<a id="generalized-azure-blob-storage-source-connector"></a>

# Generalized Azure Blob Storage Source Connector for Confluent Platform

The Generalized Kafka Connect Azure Blob Storage Source connector can read data
from any type of file naming convention listed under a Azure Blob Storage, and
the filenames don’t have to be in a specific format. As long as the files are in
any of the supported formats (for example, JSON, Avro and Byte Array) the
connector will be able to read them.

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

<a id="generalized-azure-blob-storage-source-connector-features"></a>

## Features

The Generalized Azure Blob Storage Source connector includes the following
features:

- [At least once delivery](#generalized-blob-source-at-least-once-delivery)
- [Multiple tasks](#generalized-blob-source-multiple-tasks)
- [Client-side encryption](#generalized-blob-source-csfle-sm)

<a id="generalized-blob-source-at-least-once-delivery"></a>

### At least once delivery

In the event of a task failure the connector guarantees no messages are lost,
although the last few messages may be processed again.

<a id="generalized-blob-source-multiple-tasks"></a>

### Multiple tasks

The Generalized Azure Blob Storage Source connector supports running one or more
tasks. You can specify the number of tasks in the `tasks.max` configuration
parameter. Multiple tasks may improve performance when moving a large amount of
data.

<a id="generalized-blob-source-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## Limitations

The Generalized Azure Blob Storage Source connector has the following
limitations:

- The connector won’t reload data during the following scenarios:
  - Renaming a file which the connector has already read.
  - Uploading a newer version of a file with a new record.

You should also be aware of the following connector actions:

- The connector ignores any Azure Blob Source object with a name that does not
  start with the configured `topics.dir` directory. This name is `topics/`
  by default.
- The connector uses the connector name to store offsets on how much of the
  container it has processed. Deleting a connector and using the same name will
  not cause the connector to reprocess from the beginning but will save the
  progress of the original connector unless the corresponding entry in the
  offset topic is cleared.
- For a new container, you need to create a new connector with an unused name.
  If you reconfigure an existing connector to source from the new container, or
  create a connector with a name that is used for another connector, the
  connector will not source from the beginning of data stored in the container.
  This is because the connector will maintain offsets tied to the connector
  name.

<a id="generalized-azure-blob-storage-source-connector-install"></a>

## Install the Generalized Azure Blob Storage Source 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-blob-storage-source: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-blob-storage-source:3.1.0
  ```

### Install the connector manually

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

<a id="generalized-azure-blob-storage-source-connector-license"></a>

## 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.md#azure-blob-storage-source-connector-license-config) for license
properties and [Confluent license properties](https://docs.confluent.io/kafka-connect-azure-blob-storage-sink/current/configuration_options.html#confluent-license-properties)
for information about the license topic.

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for Azure Blob Storage Source Connector for Confluent Platform](../configuration.md#azure-blob-storage-source-configuration-options).

<a id="generalized-azure-blob-storage-source-connector-quickstart"></a>

## Quick Start

In the following scenario, the Azure Blob Storage Source connector reads all
data listed under a specific container and then loads them into a Kafka topic.
You can use any file naming convention when writing data to the Azure Blob
Storage container.

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

1. Upload the following data under a folder named `quickstart` within the targeted
   Azure Blob Storage container. In this example, JSON format is used, which
   supports the following: line-delimited JSON, concatenated JSON, and a JSON
   array of records.
   ```bash
   {"f1": "value1"}
   {"f1": "value2"}
   {"f1": "value3"}
   {"f1": "value4"}
   {"f1": "value5"}
   {"f1": "value6"}
   {"f1": "value7"}
   {"f1": "value8"}
   {"f1": "value9"}
   ```
2. Install the connector using the [Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html)
   by running the following command from your Confluent Platform installation directory:
   ```properties
   confluent connect plugin install confluentinc/kafka-connect-azure-blob-storage-source:latest
   ```
3. Create a `quickstart-azureblobstoragesource.properties` file with the
   following contents:
   ```properties
   name=quickstart-azure-blob-storage-source
   connector.class=io.confluent.connect.azure.blob.storage.AzureBlobStorageSourceConnector
   tasks.max=1
   value.converter=org.apache.kafka.connect.json.JsonConverter
   mode=GENERIC
   topics.dir=quickstart
   format.class=io.confluent.connect.azure.blob.storage.format.json.JsonFormat
   topic.regex.list=quick-start-topic:.*
   azblob.account.name=my-azure-blob-storage-account
   azblob.account.key=my-azure-blob-storage-key
   azblob.container.name=my-azure-blob-storage-container
   value.converter.schemas.enable=false
   ```
4. Load the Generalized Azure Blob Storage Source connector.
   ```bash
   confluent local services connect connector load quickstart-azure-blob-storage-source --config quickstart-azure-blob-storage-source-generalized.properties
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.
5. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local services connect connector status quickstart-azure-blob-storage-source
   ```
6. Confirm that the messages are being sent to Kafka.
   ```bash
   kafka-console-consumer \
       --bootstrap-server localhost:9092 \
       --topic quick-start-topic \
       --from-beginning
   ```
7. The response should be 9 records as follows.
   ```bash
   {"f1": "value1"}
   {"f1": "value2"}
   {"f1": "value3"}
   {"f1": "value4"}
   {"f1": "value5"}
   {"f1": "value6"}
   {"f1": "value7"}
   {"f1": "value8"}
   {"f1": "value9"}
   ```

<a id="generalized-azure-blob-storage-source-connector-data-formats"></a>

## Generalized Azure Blob Storage Source Connector Data Formats

The Generalized Azure Blob Storage Source connector can read different file
formats in Azure Blob and serialize them into Kafka records. This is controlled by
the connector’s `format.class` configuration property, which has several
options:

- **Avro**: Use
  `format.class=io.confluent.connect.cloud.storage.source.format.CloudStorageAvroFormat`
  to source Avro container files.
- **JSON**: Use
  `format.class=io.confluent.connect.cloud.storage.source.format.CloudStorageJsonFormat`
  to source JSON files. Supported JSON formats are line-delimited JSON, record
  separator-limited JSON, and concatenated JSON.
- **Raw Bytes**: Use
  `format.class=io.confluent.connect.cloud.storage.source.format.CloudStorageByteArrayFormat`
  to parse the Azure Blob Storage object content as raw bytes. The default line
  separator will be the newline character, but this can be customized with the
  `format.bytearray.separator` configuration property.

## Troubleshooting Connector and Task Failures

### Stack trace

You can use the Connect [Kafka Connect REST
Interface](/platform/current/connect/references/restapi.html) to check the status of the connectors
and tasks. If a task or connector has failed, the `trace` field will include a
reason and a stack trace.

### Stack Trace message: No new files ready after scan task…

If this message is displayed, complete the following steps:

1. Review your `topics.dir` configuration property to ensure you have
   configured the right folder under the Azure Blob Storage container. If you do
   not set this parameter, the connector expects the data to be under the
   default folder which is `topics`.
2. Review the `topic.regex` configuration property to ensure your expression
   matches your data in the Azure Blob Storage container to the Kafka topic.