<a id="azure-service-bus-source-connector"></a>

# Azure Service Bus Source Connector for Confluent Platform

The Azure Service Bus Source connector 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](#servicebus-record-schema) section.

#### IMPORTANT
<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
[Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
are no longer available for download.

## Features

- [At least once delivery](#service-bus-source-at-least-once-delivery)
- [Multiple tasks](#service-bus-source-multiple-tasks)
- [No ordering guarantees](#service-bus-source-no-ordering-guarantees)
- [Fetch multiple messages](#service-bus-source-fetch-multiple-messages)
- [AMQP protocol](#service-bus-source-amqp-protocol)

The Azure Service Bus Source connector includes the following features:

<a id="service-bus-source-at-least-once-delivery"></a>

### At least once delivery

The connector guarantees that messages from Azure Service Bus are delivered at least
once to the Kafka topic.

<a id="service-bus-source-multiple-tasks"></a>

### Multiple tasks

The Azure Service Bus Source connector supports running one or more tasks. You can
specify the number of tasks in the `tasks.max` configuration parameter. This
can lead to performance gains when multiple files need to be parsed.

<a id="service-bus-source-no-ordering-guarantees"></a>

### No ordering guarantees

Although messages from different partitions may be received in a different
order, messages within a partition are guaranteed to be delivered in the same
order as they are delivered to the Azure Service Bus topic. Hence, there are ordering
guarantees within a partition.

<a id="service-bus-source-fetch-multiple-messages"></a>

### Fetch multiple messages

In every poll cycle, the connector fetches the number of messages defined by the
value set in the `azure.servicebus.max.message.count` property. The default
value is 10. However, this value can be altered depending upon the size of the
message.

<a id="service-bus-source-amqp-protocol"></a>

### 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.

## Limitations

- This connector does not currently support reading messages from [sessioned Azure Service Bus entities](https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-sessions).
- The connector does not currently support proxy configuration.

## License

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

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

See [Confluent Platform license](configuration_options.md#azure-servicebus-source-connector-license-config) for license
properties and [Confluent license properties](configuration_options.md#servicebus-source-license-topic-configuration) for information about the
license topic.

## Linux on IBM Z (s390x) support

Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x).
The connector supports the same capability available on x86_64 unless
otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support).

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for Azure Service Bus Source Connector for Confluent Platform](configuration_options.md#servicebus-source-connector-config).

## Install the Azure Service Bus Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later.
- Java 1.8.
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-azure-service-bus:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-azure-service-bus:1.3.0
  ```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-azure-service-bus)
for your connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## Quick Start

This quick start uses the Azure Service Bus Source connector to read messages from
Azure Service Bus and write them to a Kafka topic. Before you start, use the  [Azure
Service Bus Quickstart](https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-quickstart-cli)
to create a `basicqueue` queue in Azure Service Bus.

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

### Start Confluent

Start the Confluent services using the following [Confluent
CLI](https://docs.confluent.io/confluent-cli/current/index.html) command:

```bash
confluent local start
```

#### IMPORTANT
Do not use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production
environments.

### Property-based example

Create a configuration file `ServiceBusSourceConnector.properties`. This
configuration is used typically along with [standalone
workers](/platform/current/connect/concepts.html#standalone-workers).

#### IMPORTANT
Append EntityPath=<your-queue-name> at the end of the `azure.servicebus.connection.string`

```none
name=ServiceBusSourceConnector
connector.class=io.confluent.connect.azure.servicebus.ServiceBusSourceConnector
tasks.max=1
kafka.topic=servicebus-topic
azure.servicebus.sas.keyname=sas-keyname
azure.servicebus.sas.key=sas-key
azure.servicebus.namespace=servicebus-namespace
azure.servicebus.entity.name=queue-name
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.

```bash
confluent local load ServiceBusSourceConnector --config ServiceBusSourceConnector.properties
```

Confirm that the connector is in a `RUNNING` state.

```bash
confluent local status ServiceBusSourceConnector
```

### REST-based example

Use this setting with [distributed
workers](/platform/current/connect/concepts.html#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](/platform/current/connect/references/restapi.html)

#### IMPORTANT
Append EntityPath=<your-queue-name> at the end of the `azure.servicebus.connection.string`

```json
{
  "name" : "ServiceBusSourceConnector",
  "config" : {
    "connector.class" : "io.confluent.connect.azure.servicebus.ServiceBusSourceConnector",
    "tasks.max" : "1",
    "kafka.topic" : "servicebus-topic",
    "azure.servicebus.sas.keyname":"sas-keyname",
    "azure.servicebus.sas.key":"sas-key",
    "azure.servicebus.namespace":"namespace",
    "azure.servicebus.entity.name":"queue-name",
    "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).

```bash
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.

```bash
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  [Send and receive
messages](https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-quickstart-cli#send-and-receive-messages).

```bash
java -jar ./target/queuesgettingstarted-1.0.0-jar-with-dependencies.jar -c "Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=<keyName>;SharedAccessKey=<SharedAccessKey>;"
```

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

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

<a id="servicebus-record-schema"></a>

## 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.                                                                                    |