Kafka Connect Azure Service Bus Source Connector

Azure Service Bus is a multi-tenant cloud messaging service you can use to send information between applications and services. The Azure Service Bus Source connector reads data from a Azure Service Bus queue or topic and persists the data in a Kafka topic. The schema for Kafka record key and value is described in the Record Schema section.

Features

The Azure Service Bus Source connector offers the following features:

  • Atleast Once Delivery: The connector guarantees that messages from Azure Service Bus are delivered at least once to the Kafka topic.
  • No Ordering Guarantees: It is possible that the records written to Kafka topic end up in a different order as compared to Service Bus message-entity.
  • Fetch Multiple Messages In every poll cycle, the connector fetches azure.servicebus.max.message.count number of messages. By default, this value is 10. However, this can be altered depending upon the size of the message.
  • AMQP Protocol This connector is based on the AMQP protocol so it should work with other servers that implement this protocol.

Note

While creating the queue for the Service Bus queue or topic, the lock duration should be set to a high enough value to avoid duplicating records in Kafka topic. This allows the connector to commit the records and send acknowledgement for each Service Bus message processed. In the case when the connector fails to write records to Kafka topic, the messages in the Service Bus topic are made available again.

Prerequisites

The following are required to run the Kafka Connect Azure Service Bus Sink Connector:

  • Kafka Broker: Confluent Platform 3.3.0 or above, or Kafka 0.11.0 or above
  • Connect: Confluent Platform 4.0.0 or above, or Kafka 1.0.0 or above
  • Java 1.8

License

You can use this connector for a 30-day trial period without a license key.

After 30 days, this connector is available under a Confluent enterprise license. Confluent issues enterprise license keys to subscribers, along with providing enterprise-level support for Confluent Platform and your connectors. If you are a subscriber, please contact Confluent Support at support@confluent.io for more information.

See Confluent Platform license for license properties and License topic configuration for information about the license topic.

Install Azure Service Bus Connector

You can install this connector by using the Confluent Hub client (recommended) or you can manually download the ZIP file.

Install the connector using Confluent Hub

Prerequisite
Confluent Hub Client must be installed. This is installed by default with Confluent Enterprise.

Navigate to your Confluent Platform installation directory and run this command to install the latest (latest) connector version. The connector must be installed on every machine where Connect will be run.

confluent-hub install confluentinc/kafka-connect-servicebus:latest

You can install a specific version by replacing latest with a version number. For example:

confluent-hub install confluentinc/kafka-connect-servicebus:1.0.0-preview

Quick Start

This quick start uses the Azure Service Bus Source Connector to read messages from Azure Service Bus and write them to Kafka topic.

Start Confluent

Start the Confluent services using the following Confluent CLI command:

confluent start

Important

Do not use the Confluent CLI in production environments.

Property-based example

Create a configuration file for the connector. This file is included with the connector in etc/kafka-connect-azure-servicebus/ServiceBusSourceConnector.properties. This configuration is used typically along with standalone workers.

name=ServiceBusSourceConnector
connector.class=io.confluent.connect.azure.servicebus.ServiceBusSourceConnector
tasks.max=1
kafka.topic=servicebus-topic
azure.servicebus.msg.entity.type=queue
azure.servicebus.connection.string=Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=keyName;SharedAccessKey=key;EntityPath=queueName
azure.servicebus.subscription=""
azure.servicebus.max.message.count=10
azure.servicebus.max.waiting.time.seconds=30
confluent.license=
confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1

Run the connector with this configuration.

confluent load ServiceBusSourceConnector -d ServiceBusSourceConnector.properties

Confirm that the connector is in a RUNNING state.

confluent status ServiceBusSourceConnector

REST-based example

Use this setting with distributed workers. Write the following JSON to config.json, configure all of the required values, and use the following command to post the configuration to one of the distributed connect workers. Check here for more information about the Kafka Connect REST API

{
  "name" : "ServiceBusSourceConnector",
  "config" : {
    "connector.class" : "io.confluent.connect.azure.servicebus.ServiceBusSourceConnector",
    "tasks.max" : "1",
    "kafka.topic" : "servicebus-topic",
    "azure.servicebus.msg.entity.type" : "queue",
    "azure.servicebus.connection.string" : "Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=keyName;SharedAccessKey=key;EntityPath=queueName",
    "azure.servicebus.subscription" : "",
    "azure.servicebus.max.message.count" : "10",
    "azure.servicebus.max.waiting.time.seconds" : "30",
    "confluent.license":"",
    "confluent.topic.bootstrap.servers":"localhost:9092",
    "confluent.topic.replication.factor":"1"
  }
}

Use curl to post the configuration to one of the Kafka Connect Workers. Change http://localhost:8083/ the endpoint of one of your Kafka Connect worker(s).

curl -s -X POST -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors

Use the following command to update the configuration of existing connector.

curl -s -X PUT -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors/ServiceBusSourceConnector/config

To publish messages to Service Bus queue, follow the Azure Service Bus Quickstart.

To consume records written by connector to the configured Kafka topic, run the following command:

kafka-avro-console-consumer \
--bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 \
--topic servicebus-topic \
--from-beginning

Record Schema

Then source connector creates records in the following format:

Key Schema

The Key is a struct with the following fields:

Field Name Schema Type Description
MessageId String The message identifier that uniquely identifies the message and its payload.

Value Schema

The Value is a struct with the following fields:

Field Name Schema Type Description
deliveryCount int64 The number of the times this message was delivered to clients.
enqueuedTimeUtc int64 The time at which this message was enqueued in Azure Service Bus.
contentType String The content type of this message.
label String The application specific message label.
correlationId Optional String The correlation identifier.
messageProperties Optional String The map of user application properties of this message.
partitionKey Optional String The partition key for sending a message to a partitioned entity.
replyTo Optional String The address of an entity to send replies to.
replyToSessionId Optional String The session identifier augmenting the ReplyTo address.
deadLetterSource Optional String The name of the queue or subscription that this message was enqueued on, before it was deadlettered.
timeToLive int64 The duration before this message expires.
lockedUntilUtc Optional int64 The time when the lock of this message expires.
sequenceNumber Optional int64 The unique number assigned to a message by Azure Service Bus.
sessionId Optional String The session identifier for a session-aware entity.
lockToken Optional String The lock token for the current message.
messageBody bytes The body of this message as a byte array.
getTo Optional String The “to” address.

Additional Documentation