1 of 78

Kafka Connectors

Overview

Lenses Kafka Connectors are a collection of open-source, Apache 2.0 licensed, Kafka Connect Connectors. Maintained by Lenses.io to deliver open-source Kafka Connectors to the community.

Understanding Kafka Connect

This page describes an overview of Kafka Connect.

Kafka Connect uses the Kafka Producer API and the Kafka Consumer API to load data into Apache Kafka and output data from Kafka to another storage engine. A Connector is an instantiation of a connector plugin defined by a configuration file.

Kafka Connect Core Concepts

Kafka Connect is a plugin framework. It exposes APIs for developers to implement standard recipes (known as Connector Plugins) for exchanging data between Kafka and other data systems. It provides the runtime environment to execute the plugins in the form of Connect Clusters, made up of Workers.

The fundamental building blocks of Connect Clusters are the Workers. Workers manage the plumbing required to interact with Kafka, which serves to coordinate the workload. They also handle the management of connector plugins and the creation and supervision of their instances, known as connector instances.

Connect Clusters can be configured either as a standalone cluster (consisting of one Worker) or distributed (consisting of many Workers, ideally on different machines). The benefit of the distributed mode is fault tolerance and load balancing of workload across machines.

A Connector Plugin is a concrete implementation of the Plugin APIs exposed by Connect for a specific third-party system, for example, S3. If a Connector Plugin is extracting data from a system and writing to Kafka, it’s called a Source Plugin. If it ships data from Kafka to another system, it’s called a Sink Plugin.

The modular architecture of Connect, with Workers managing connector plugins and connector instances, abstracts away the complexities of data integration, enabling developers to concentrate on the core functionality of their applications.

Connectors Plugins

The Community has been developing Connectors to connect to all sorts of systems for years. Here at Lenses, we are the biggest contributor to open-source connectors via our .

The Connector Plugins are code, deployed as jar files, and added to the classpath of each Worker in the Connect cluster.

Why does it need to be on each Worker? The reason lies in the distributed nature of Connect and its workload distribution mechanism. The Connect framework evenly distributes tasks among multiple Workers using the Kafka consumer group protocol. This ensures load balancing and fault tolerance within the Connect cluster. Each Worker is responsible for executing a subset of the assigned tasks.

To enable the execution of assigned workloads, each Worker needs to have access to the required plugins locally. By having the plugins available on each Worker's classpath, the Connect framework can dynamically load and utilize them whenever necessary. This approach eliminates the need for plugins to be distributed separately or retrieved from a centralized location during runtime. But it does mean that when you install your Connect Cluster that you also need to ensure each plugin is installed.

Tasks vs Connectors

Each Connector Plugin has to implement two interfaces:

Connector (Source or Sink)
Task (Source or Sink)

The Connector interface is responsible for defining and configuring the instance of the Connector, for example, validating the configuration and then splitting that configuration up for the Connect APIs to distribute amongst the workers.

The actual work is done by the Task class. Connect sends out, via Kafka, a configuration, defined by the Connector class, to each worker. Each assigned Worker picks up the task (it's listening to the configuration topic) and creates an instance of the task.

Connector instances are created on the Worker you submit the creation request to, task instances can also be on the same Worker, but also on the other Workers. Connect distributes the tasks to Workers via Kafka; they have internal consumer groups listening to the system topics Connect uses. If you look into the default topic, connect-configs, you will see the split of the Connector configs for each task.

Converters

A Kafka message is made up of:

Headers
Key
Value

Headers are a map, while the key and value are stored as byte arrays in Kafka. Typically, these byte arrays represent data stored as JSON or AVRO, however, it could be anything.

To decouple from the format of data inside Kafka topics, Connect uses an internal format called . Structs have fields, and fields have types defined in a Schema.

Converters translate the byte arrays, which could be AVRO or other formats, to Struct for Sink connectors and vice versa for Source connectors.

For example, you use Avro as your data format in your cluster. The Avro converter allows you to build connectors and interact with Connect Struct only; you may decide to move to Protobuf later, and don't want to reimplement your connector. This is where the converter comes in. It handles converting the Struct to Avro for Source connectors and Struct to external systems, e.g. Cassandra, for sinks. You can then swap this out for a different converter; you, as a developer, only deal with Structs.

The following converters are available with Apache Kafka:

org.apache.kafka.connect.converters.DoubleConverter: Serializes to and deserializes from DOUBLE values. When converting from bytes to Connect format, the converter returns an optional FLOAT64 schema.
org.apache.kafka.connect.converters.FloatConverter: Serializes to and deserializes from FLOAT values. When converting from bytes to Connect format, the converter returns an optional FLOAT32 schema.

Confluent provides support for (Schema Registry required):

AvroConverter io.confluent.connect.avro.AvroConverter
ProtobufConverter io.confluent.connect.protobuf.ProtobufConverter

Secret Providers

Kafka Connect supports an internal data type of Secret. If a connector implements this type as part of its Config definition, it will be masked in any logging, however, it will still be exposed in API calls to Kafka Connect Workers.

To solve this, Kafka Connect supports secret provider plugins. They allow for indirect references, resolved at the initialization of a Connector instance, to external secret providers.

Lenses.io provides Secret Providers for Azure, AWS, Hashicorp Vault and Environment variables .

Single Message Transforms (SMTs)

SMTs are plugins that enable users to manipulate records, one at a time. For Source Connectors they manipulate records after the Task has handed them back to the Connect Framework and before they are written to Kafka. For Sink Connectors, they allow for manipulation of records before they are passed from the Connect Framework to the Sink Task.

Apache Kafka comes with several available SMTs, for which the documentation can be found .

Lenses.io also provides a number of SMTs, which can be found .

Connectors

This page describes the Stream Reactor connector plugins.

Install

This page describes installing the Lenses Kafka Connectors.

If you do not use the plugin.path and add the connectors directly to the CLASSPATH you may have dependency conflicts.

Download the release and unpack.

Within the unpacked directory you will find the following structure:

/opt/stream-reactor-x.x.x-x.x.x
├── bin
├── conf
├── libs
├── LICENSE

The libs directory contains all the Stream Reactor Connector jars. Edit your Connect worker properties add the path to the directory containing the connectors and restart your workers. Repeat this process for all the Connect workers in your cluster. The connectors must be available to all the workers.

Example:

Sources

This page details the configuration options for the Stream Reactor Kafka Connect source connectors.

Source connectors read data from external systems and write to Kafka.

Sinks

This page details the configuration options for the Stream Reactor Kafka Connect sink connectors.

Sink connectors read data from Kafka and write to an external system.

FAQ

Can the datalakes sinks lose data?

Kafka topic retention policies determine how long a message is retained in a topic before it is deleted. If the retention period expires and the connector has not processed the messages, possibly due to not running or other issues, the unprocessed Kafka data will be deleted as per the retention policy. This can lead to significant data loss since the messages will no longer be available for the connector to sink to the target system.

Do the datalake sinks support exactly once semantics?

Yes, the datalakes connectors natively support exactly-once guarantees.

How do I escape dots in field names in KCQL?

Field names in Kafka message headers or values may contain dots (.). To access these correctly, enclose the entire target in backticks (```) and each segment which consists of a field name in single quotes ('):

How do I escape other special characters in field names in KCQL?

For field names with spaces or special characters, use a similar escaping strategy:

Field name with a space: `_value.'full name'`
Field name with special characters: `_value.'$special_characters!'`

This ensures the connector correctly extracts the intended fields and avoids parsing errors.

Azure Event Hubs

This page describes the usage of the Stream Reactor Azure Event Hubs Sink Connector.

Coming soon!

Secret Providers

This page describes how to install secret provider plugins in your Kafka Connect Clusters.

Install

This page describes install the Lenses Secret Providers for Kafka Connect.

Add the plugin to the worker classloader isolation via the plugin.path option:

plugin.path=/usr/share/connectors,/opt/secret-providers

For Azure do not use the classloader isolation (plugin.path) that Kafka Connect provides. The Azure SDK uses the default classloader and will not find the HTTP client.

Secret Rotation

To allow for secret rotation add config.action.reload to your Connect workers properties files.

This property accepts one of two options:

none - No action happens at a connector failure (e.g. it can no longer connect to an external system)
restart - The work will schedule a restart of the connector

Secrets will only be reloaded if the Connector restarts.

Environment Provider

This page describes how to retrieve secrets from Environment variables for use in Kafka Connect.

Use Environment variables to hold secrets and use them in Kafka Connect.

Secrets will only be reloaded if the Connector restarts.

Configuration

Example Worker Properties:

Usage

To use this provider in a connector, reference the ENVSecretProvider environment variable providing the value of the connector property.

The indirect reference is in the form ${provider::key} where:

provider is the name of the provider in the worker property file set above
key is the name of the environment variable holding the secret.

For example, if we store two secrets as environment variables:

MY_ENV_VAR_USERNAME with the value lenses and
MY_ENV_VAR_PASSWORD with the value my-secret-password

we would set:

This would resolve at runtime to:

Data encoding

This provider inspects the value of the environment to determine how to process the value. The value can optionally provide value metadata to support base64 decoding and writing values to files.

To provide metadata the following patterns are expected:

where value is the actual payload and metadata can be one of the following:

ENV-base64 - the provider will attempt to base64 decode the value string
ENV-mounted-base64 - the provider will attempt to base64 decode the value string and write to a file
ENV-mounted - the provider will write the value to a file

if no metadata is found the value of the environment variable is returned.

AES256

Decodes values encoded with AES-256 to enable passing encrypted values to connectors.

Secrets will only be reloaded if the Connector restarts.

Add the plugin to the worker classloader isolation via the plugin.path option:

The provider gets AES-256 encrypted value as a key and simply decrypts it to get the value (instead of e.g. looking up for the value somewhere).

The AES-256 encryption used for the value needs to be prefixed with base64 encoded initialisation vector and a space character, the encrypted value is also base64 encoded. So to corretly encrypt value1 I need to follow following steps:

encrypted-bytes = aes-256 encrypted value1
encrypted-base64 = base64 encrypted-bytes

Configuring the plugin

The plugin needs to be configured with secret key that will be used for decoding. The key is a string and needs to have size of 32 bytes (UTF-8 encoded).

Name

Description

Default

Example worker properties file:

Usage

To use this provider in a connector, reference the keyvault containing the secret and the key name for the value of the connector property.

The indirect reference is in the form ${provider:path:key} where:

provider is the name of the provider in the worker property file set above
path used to provide encoding of the value: utf8, utf8_file, base64, base64_file
key is the AES-256 encrypted value to be decrypted by the plugin

For example, if hello aes-256 encrypted using some key equals to xyxyxy - then if I configure connector to use ${aes256::xyxyxy} for a parameter value, the value should be substituted with “hello” string:

This would resolve at runtime to:

path belonging to key reference is used to specify encoding used to pass the value. The provider supports following encodings:

base64: base-64 encoding of the textual value
base64_file: base-64 encoding of the value that when decrypted should be stored in the file
utf8_file: utf-8 encoding of the value that when decrypted should be stored in the file

The UTF8 means the value returned is the decrypted value of the encrypted value (key). The BASE64 means the value returned is the base64 decoded decrypted value of the encrypted value (key).

If the value for the encoding is UTF8_FILE the string contents are written to a file. The name of the file will be randomply generated. The file location is determined by the file.dir configuration option given to the provider via the Connect worker.properties file.

If the value for the encoding is BASE64_FILE the string contents are based64 decoded and written to a file. The name of the file will be randomply generated. For example, if a connector needs a PEM file on disk, set this as the path as BASE64_FILE. The file location is determined by the file.dir configuration option given to the provider via the Connect worker.properties file.

If the key reference path is not set or is set to unknown value - utf8 encoding is used as default.

For example, if we want to save hi there ! to the file, and aes-256 encrypted content equals xyxyxy - then if I configure connector to use ${aes256:utf8_file:xyxyxy} for a parameter value, the provider will create new file with random name (abc-def-ghi) and store hi there ! to the file. If configured store directory is /store-root, he value will be substituted with /store-root/secrets/abc-def-ghi string:

resolves to

Single Message Transforms

This page describes how to used SMTs your Kafka Connect Clusters.

Overview

This page describes an overview of the Lenses SMTs for Kafka Connect.

Lenses provides several SMTs designed for use with Stream Reactor connectors, you can also use them with other connectors or your own.

These SMTs are designed to be used with the Kafka Connect framework. The SMTs create record headers. The advantage of using headers is that they reduce the memory and CPU cycles required to change the payload. See for example the Kafka Connect TimestampConverter. Furthermore, they support Stream-Reactor S3 sink partitioners, for scenarios like:

Partitioning by the system clock (e.g. using the system clock as a partition key with a yyyy-MM-dd-HH format)
Partitioning by a rolling window (e.g. every 15 minutes, or one hour)
Partitioning by a custom timestamp (e.g. a timestamp field in the payload, record Key or Value)
Partitioning by a custom timestamp with a rolling window (e.g. a timestamp field in the payload, every 15 minutes, or one hour)

Installation

SMT installation on Lenses platform

Add the plugin to the worker classloader isolation via the plugin.path option:

SMT installation on MSK connect

For MSK connect you need to bundle your SMT with the connector you need to use and deploy as one plugin.

To do so, download your connector to get the jar (unzip if needed), download (unzip it). Put both jar in the same folder, and zip them altogether.

This zip (containing both jars at the same level) must be uploaded as a plugin in MSK connect.

InsertRecordTimestampHeaders

SMT that inserts date, year, month, day, hour, minute and second headers using the record timestamp. If the record timestamp is null, the SMT uses the current system time.

The headers inserted are of type STRING. By using this SMT, you can partition the data by yyyy-MM-dd/HH or yyyy/MM/dd/HH, for example, and only use one SMT.

The list of headers inserted are:

date

InsertRollingWallclock

SMT that inserts the system clock value as a message header, a value adapted to a specified time window boundary, for example every 15 minutes, or one hour.

The value inserted is stored as a STRING, and it holds either a string representation of the date and time epoch value, or a string representation of the date and time in the format specified.

Transform Type Class

InsertSourcePartitionOrOffsetValue

The InsertSourcePartitionOrOffsetValue transformation in Kafka Connect allows you to insert headers into SourceRecords based on partition or offset values.

The InsertSourcePartitionOrOffsetValue transformation in Kafka Connect allows you to insert headers into SourceRecords based on partition or offset values. This is useful for adding metadata to your data records before they are sent to destinations like AWS S3, Azure Datalake, or GCP Storage.

This SMT only works with source connectors.

InsertWallclock

Inserts the system clock as a message header.

Use SMT if you want to use more than one date time part. This avoids multiple SMTs and is more efficient.

For example, if you want to partition the data by yyyy-MM-dd/HH, then you can use InsertWallclockHeaders which inserts multiple headers: date, year, month, day, hour, minute, second.

InsertWallclockDateTimePart

A Kafka Connect Single Message Transform (SMT) that inserts the system clock year, month, day, minute, or seconds as a message header, with a value of type STRING.

Use InsertWallclockHeaders SMT if you want to use more than one date time part. This avoids multiple SMTs and is more efficient.

For example, if you want to partition the data by yyyy-MM-dd/HH, then you can use InsertWallclockHeaders which inserts multiple headers: date, year, month, day, hour, minute, second.

Transform Type Class

Configuration

Name

Description

Type

Default

Valid Values

Importance

Example

To store the year, use the following configuration:

To store the month, use the following configuration:

To store the day, use the following configuration:

To store the hour, use the following configuration:

To store the hour, and apply a timezone, use the following configuration:

To store the minute, use the following configuration:

To store the second, use the following configuration:

Tutorials

This page contains tutorials for common Kafka Connect use cases.

Creating & managing a connector

This page describes managing a basic connector instance in your Connect cluster.

Creating your Kafka Connector Configuration

To deploy a connector into the Kafka Connect Cluster, you must follow the steps below:

You need to have the Jars in your Kafka Connect Cluster

Cloud Storage Examples

Examples for AWS S3, Azure Datalake and GCP Storage connectors.

Http Sink Templating

The page provides examples of HTTP Sink templating.

Static string template

In this case the converters are irrelevant as we are not using the message content to populate our message template.

Sink converters & different data formats

This page describes configuring sink converters for Kafka Connect.

You can configure the converters either at the Connect worker level or at the Connector instance level.

Schema.Struct and a Struct Payload

If you follow the best practice while producing the events, each message should carry its schema information. The best option is to send AVRO.

This requires the SchemaRegistry.

Schema.String and a JSON Payload

Sometimes the producer would find it easier to just send a message with Schema.String and a JSON string. In this case, your connector configuration should be set to value.converter=org.apache.kafka.connect.json.JsonConverter. This doesn’t require the SchemaRegistry.

No schema and a JSON Payload

Many existing systems publish JSON over Kafka and bringing them in line with best practices is quite a challenge, hence we added the support. To enable this support you must change the converters in the connector configuration.

Source converters with incoming JSON or Avro

This page describes how to use converters with source systems sending JSON and Avro.

Source converters depend on the source system you are reading data from. The Connect SourceTask class requires you to supply a List of SourceRecords. Those records can have a schema but how the schema is translated from the source system to a Connect Struct depends on the connector.

We provide four converters out of the box but you can plug in your own. The WITHCONVERTER keyword supports this option. These can be used when source systems send records as JSON or AVRO, for example, MQTT or JMS.

Loading XML from Cloud storage

Coming soon!

Loading ragged width files

Coming soon!

Using the MQTT Connector with RabbitMQ

Coming soon!

Using Error Policies

This page describes how to use Error policies in Stream Reactor sink connectors.

In addition to the dead letter queues provided by Kafka Connect, the Stream Reactor sink connectors support Error Policies to handle failure scenarios.

The sinks have three error policies that determine how failed writes to the target database are handled. These error policies allow you to control the behaviour of the sink if it encounters an error when writing records to the target system. Since Kafka retains the records, subject to the configured retention policy of the topic, the sink can ignore the error, fail the connector or attempt redelivery.

Name

Description

Default Value

Using dead letter queues

Coming soon!

Contributing

This page describes how to contribute to the Stream Reactor.

The Stream Reactor is an open-source project and we welcome feedback on any issues, even more, we enjoy and encourage contributions, either fixes or new connectors! Don't be shy.

What you can contribute? Everything, even if you aren't a JVM program you can help with docs and tutorials, everything is open.

How to contribute?

If you have an issue raise it on the GitHub issue page and supply as much information as you can. If you have a fix, raise a Pull request and ask for a review.

Please make sure to:

Test your changes, we won't accept changes without unit and integration testing
Make sure it builds!
Rebase your issues if they become stale

The Stream Reactor is under an Apache 2.0 license so others can use your hard work for free without restriction.

Developing a connector

This section describes how to contribute a new connector to the Stream Reactor.

SBT, Java 11, and Scala 2.13 are required.

Setting up a new module

Utilities

Coming soon!

Testing

Coming soon!

Downloads

GitHub Release downloads for Stream Reactor Connectors, Secret Providers and SMTs.

Release notes

This page describes the release notes for the Stream Reactor.

Secret Providers

This page contains the release notes for Connect Secret Providers.

2.3.0

Security: Write maven Descriptors on packaging to avoid incorrect dependencies being identified by security scanner tools. (Fixes CVE-2023-1370).
Security: Add dependency checking as part of the build process.

AES256 Provider:

Security: Change AES256 key to PASSWORD type to avoid logging secrets.

AWS Secrets Manager Provider:

New property : file.write Writes secrets to file on path. Required for Java trust stores, key stores, certs that need to be loaded from file. For ease of use for the secret provider, this is disabled by default.
New property : secret.default.ttl If no TTL is configured in AWS Secrets Manager, apply a default TTL (in milliseconds).

Azure Secret Provider:

Bugfix: Recompute TTL values on each get so the timestamp of reschedule shrinks until TTL is reached.
Bugfix: Fix so that UTF-8 encodings in Azure are correctly mapped to the UTF8 encoding in the secret provider.

Hashicorp Vault Provider:

Bugfix: Files will be written to the correct directory.
New property: app.role.path Support vault approle custom mount path.
New property: kubernetes.auth.path Support vault custom auth path (with default value to be auth/kubernetes).

Understanding Kafka Connect

This page describes an overview of Kafka Connect.

Kafka Connect Core Concepts

Connectors Plugins

The Community has been developing Connectors to connect to all sorts of systems for years. Here at Lenses, we are the biggest contributor to open-source connectors via our .

The Connector Plugins are code, deployed as jar files, and added to the classpath of each Worker in the Connect cluster.

Tasks vs Connectors

Each Connector Plugin has to implement two interfaces:

Connector (Source or Sink)
Task (Source or Sink)

Converters

A Kafka message is made up of:

Headers
Key
Value

Headers are a map, while the key and value are stored as byte arrays in Kafka. Typically, these byte arrays represent data stored as JSON or AVRO, however, it could be anything.

To decouple from the format of data inside Kafka topics, Connect uses an internal format called . Structs have fields, and fields have types defined in a Schema.

Converters translate the byte arrays, which could be AVRO or other formats, to Struct for Sink connectors and vice versa for Source connectors.

The following converters are available with Apache Kafka:

org.apache.kafka.connect.converters.DoubleConverter: Serializes to and deserializes from DOUBLE values. When converting from bytes to Connect format, the converter returns an optional FLOAT64 schema.
org.apache.kafka.connect.converters.FloatConverter: Serializes to and deserializes from FLOAT values. When converting from bytes to Connect format, the converter returns an optional FLOAT32 schema.

Confluent provides support for (Schema Registry required):

AvroConverter io.confluent.connect.avro.AvroConverter
ProtobufConverter io.confluent.connect.protobuf.ProtobufConverter

Secret Providers

To solve this, Kafka Connect supports secret provider plugins. They allow for indirect references, resolved at the initialization of a Connector instance, to external secret providers.

Lenses.io provides Secret Providers for Azure, AWS, Hashicorp Vault and Environment variables .

Single Message Transforms (SMTs)

Apache Kafka comes with several available SMTs, for which the documentation can be found .

Lenses.io also provides a number of SMTs, which can be found .

Azure Service Bus

This page describes the usage of the Stream Reactor Azure Service Bus Sink Connector.

Stream Reactor Azure Service Bus Sink Connector is designed to effortlessly translate Kafka records into your Azure Service Bus cluster. It leverages Microsoft Azure API to transfer data to Service Bus in a seamless manner, allowing for their safe transition and safekeeping both payloads and metadata (see Payload support). It supports both types of Service Buses: Queues and Topics. Azure Service Bus Source Connector provides its user with AT-LEAST-ONCE guarantee as the data is committed (marked as read) in Kafka topic (for assigned topic and partition) once Connector verifies it was successfully committed to designated Service Bus topic.

Connector Class

Full Config Example

For more examples see the .

The following example presents all the mandatory configuration properties for the Service Bus connector. Please note there are also optional parameters listed in . Feel free to tweak the configuration to your requirements.

KCQL support

You can specify multiple KCQL statements separated by ; to have the connector map between multiple topics.

The following KCQL is supported:

It allows you to map Kafka topic of name <your-kafka-topic> to Service Bus of name <your-service-bus> using the PROPERTIES specified (please check for more info on necessary properties)

The selection of fields from the Service Bus message is not supported.

Authentication

You can connect to an Azure Service Bus by passing your connection string in configuration. The connection string can be found in the Shared access policies section of your Azure Portal.

Learn more about different methods of connecting to Service Bus on the .

QUEUE and TOPIC Mappings

The Azure Service Bus Connector connects to Service Bus via Microsoft API. In order to smoothly configure your mappings you have to pay attention to PROPERTIES part of your KCQL mappings. There are two cases here: reading from Service Bus of type QUEUE and of type TOPIC. Please refer to the relevant sections below. In case of further questions check to learn more about those mechanisms.

Writing to QUEUE ServiceBus

In order to be writing to the queue there's an additional parameter that you need to pass with your KCQL mapping in the PROPERTIES part. This parameter is servicebus.type and it can take one of two values depending on the type of the service bus: QUEUE or TOPIC. Naturally for Queue we're interested in QUEUE here and we need to pass it.

This is sufficient to enable you to create the mapping with your queue.

Writing to TOPIC ServiceBus

In order to be writing to the topic there is an additional parameter that you need to pass with your KCQL mapping in the PROPERTIES part:

Parameter servicebus.type which can take one of two values depending on the type of the service bus: QUEUE or TOPIC. For topic we're interested in TOPIC here and we need to pass it.

This is sufficient to enable you to create the mapping with your topic.

Disabling batching

If the Connector is supposed to transfer big messages (size of one megabyte and more), Service Bus may not want to accept a batch of such payloads, failing the Connector Task. In order to remediate that you may want to use batch.enabled parameter, setting it to false. This will sacrifice the ability to send the messages in batch (possibly doing it slower) but should enable user to transfer them safely.

For most of the usages, we recommend omitting it (it's set to true by default).

Kafka payload support

This sink supports the following Kafka payloads:

String Schema Key and Binary payload (then MessageId in Service Bus is set with Kafka Key)
any other key (or keyless) and Binary payload (this causes Service Bus messages to not have specified MessageId)
No Schema and JSON

Null Payload Transfer

Azure Service Bus doesn't allow to send messages with null content (payload)

Null Payload (sometimes referred as Kafka Tombstone) is a known concept in Kafka messages world. However, because of Service Bus limitations around that matter, we aren't allowed to send messages with null payload and we have to drop them instead.

Please keep that in mind when using Service Bus and designing business logic around null payloads!

Option Reference

KCQL Properties

Please find below all the necessary KCQL properties:

Name

Description

Type

Default Value

Configuration parameters

Please find below all the relevant configuration parameters:

Name

Description

Type

Default Value

Azure CosmosDB

This page describes the usage of the Stream Reactor Azure CosmosDB Sink Connector.

Version 10.0.0 introduces breaking changes: the connector has been renamed from DocumentDB, uses the official CosmosDB SDK, and supports new bulk and key strategies.

A Kafka Connect sink connector for writing records from Kafka to Azure CosmosDB using the SQL API.

Connector Class

Example

For more examples see the .

KCQL support

You can specify multiple KCQL statements separated by**;** to have a connector sink multiple topics. The connector properties topics or topics.regex are required to be set to a value that matches the KCQL statements.

The following KCQL is supported:

Examples:

Insert Mode

Insert is the default write mode of the sink. It inserts messages from Kafka topics into CosmosDB.

Upsert Mode

The Sink supports CosmosDB upsert functionality which replaces the existing row if a match is found on the primary keys.

This mode works with at least once delivery semantics on Kafka as the order is guaranteed within partitions. If the same record is delivered twice to the sink, it results in an idempotent write. The existing record will be updated with the values of the second which are the same.

If records are delivered with the same field or group of fields that are used as the primary key on the target table, but different values, the existing record in the target table will be updated.

Since records are delivered in the order they were written per partition the write is idempotent on failure or restart. Redelivery produces the same result.

Bulk Mode

Bulk mode enables efficient batching of writes to CosmosDB, reducing API calls and improving throughput. Enable it with:

When enabled, you can control batching behavior using the KCQL PROPERTIES clause (flush.size, flush.count, flush.interval). If disabled, records are written individually.

Custom Key Strategy

You can control how the connector populates the id field in CosmosDB documents using:

connect.cosmosdb.key.source (Key, Metadata, KeyPath, ValuePath)
connect.cosmosdb.key.path (field path, used with KeyPath or ValuePath)

This allows you to use the Kafka record key, metadata, or a specific field from the key or value as the document ID.

Error Handling

Configure error handling and retry behavior with:

connect.cosmosdb.error.policy (NOOP, THROW, RETRY)
connect.cosmosdb.max.retries (max retry attempts)
connect.cosmosdb.retry.interval (milliseconds between retries)

These settings control how the connector responds to errors during writes.

Throughput

You can set the manual throughput (RU/s) for new CosmosDB collections with:

The default is 400 RU/s, which is the minimum allowed by Azure Cosmos DB and is cost-effective for most workloads.

Proxy

If you need to connect via a proxy, specify the proxy details with:

Progress Reporting

Enable progress reporting to log how many records have been processed:

Option Reference

Name

Description

Type

Default Value

KCQL Properties

Note: The following KCQL PROPERTIES options are only available when connect.cosmosdb.bulk.enabled=true. If bulk mode is disabled, records are written individually.

The CosmosDB Sink Connector supports the following KCQL PROPERTIES options to control file flushing behavior:

Property

Description

Type

Default Value

Flush Options Explained:

Flush by Count: Triggers a flush after a specified number of records have been written.
Flush by Size: Initiates a flush once a predetermined size (in bytes) has been reached.
Flush by Interval: Enforces a flush after a defined time interval (in seconds), acting as a fail-safe to ensure timely data management even if other flush conditions are not met.

You can use these properties in your KCQL statement's PROPERTIES clause, for example:

Kafka payload support

This sink supports the following Kafka payloads:

Schema.Struct and Struct (Avro)
Schema.Struct and JSON
No Schema and JSON

Error policies

The connector supports .

Endpoint and Master Key Configuration

To connect to your Azure CosmosDB instance, the connector requires two essential configuration properties:

connect.cosmosdb.endpoint: This specifies the URI of your CosmosDB account. It should point to the connection endpoint provided in the Azure CosmosDB dashboard.
connect.cosmosdb.master.key: This is the authentication key used to access the database. Note: Azure often refers to this as the primary key, which can be confusing—the master.key in the connector configuration corresponds to Azure's primary key value.

Both properties are mandatory and must be set to establish a connection with CosmosDB.

Key Population Strategy

The connector offers flexible options for populating the id field of documents in CosmosDB. The behavior is controlled by two configurations:

connect.cosmosdb.key.source

Description: Defines the strategy used to extract or generate the document ID.
Valid Values:
- Key (default): Use the Kafka record key.

connect.cosmosdb.key.path

Description: Specifies the field path to extract the key from, when using KeyPath or ValuePath.
Default: id
Example: If using ValuePath with connect.cosmosdb.key.path=id, the connector will use value.id

HTTP

This page describes the usage of the Stream Reactor HTTP Sink Connector.

A Kafka Connect sink connector for writing records from Kafka to HTTP endpoints.

Features

Support for Json/Avro/String/Protobuf messages via Kafka Connect (in conjunction with converters for Schema-Registry based data storage).
URL, header and content templating ability give you full control of the HTTP request.
Configurable batching of messages, even allowing you to combine them into a single request selecting which data to send with your HTTP request.

Connector Class

Example

For more examples see the .

Content Template

The Lenses HTTP sink comes with multiple options for content templating of the HTTP request.

Static Templating

If you do not wish any part of the key, value, headers or other data to form a part of the message, you can use static templating:

Single Message Templating

When you are confident you will be generating a single HTTP request per Kafka message, then you can use the simpler templating.

In your configuration, in the content property of your config, you can define template substitutions like the following example:

(please note the XML is only an example, your template can consist of any text format that can be submitted in a http request)

Multiple Message Templating

To collapse multiple messages into a single HTTP request, you can use the multiple messaging template. This is automatic if the template has a messages tag. See the below example:

Again, this is an XML example but your message body can consist of anything including plain text, json or yaml.

Your connector configuration will look like this:

The final result will be HTTP requests with bodies like this:

Available Keys

When using simple and multiple message templating, the following are available:

Field

Usage Example

URL Templating

URL including protocol (eg. http://lenses.io). Template variables can be used.

The URL is also a so can contain substitutions from the message key/value/headers etc. If you are batching multiple kafka messages into a single request, then the first message will be used for the substitution of the URL.

Authentication Options

Currently, the HTTP Sink supports either no authentication, BASIC HTTP authentication and OAuth2 authentication.

No Authentication (Default)

By default, no authentication is set. This can be also done by providing a configuration like this:

BASIC HTTP Authentication

BASIC auth can be configured by providing a configuration like this:

OAuth2 Authentication

OAuth auth can be configured by providing a configuration like this:

Headers List

To customise the headers sent with your HTTP request you can supply a Headers List.

Each header key and value is also a so can contain substitutions from the message key/value/headers etc. If you are batching multiple kafka messages into a single request, then the first message will be used for the substitution of the headers.

Example:

SSL Configuration

Enabling SSL connections between Kafka Connect and HTTP Endpoint ensures that the communication between these services is secure, protecting sensitive data from being intercepted or tampered with. SSL (or TLS) encrypts data in transit, verifying the identity of both parties and ensuring data integrity. Please check out section in order to set it up.

Batch Configuration

The connector offers three distinct flush options for data management:

Flush by Count - triggers a file flush after a specified number of records have been written to it.
Flush by Size - initiates a file flush once a predetermined size (in bytes) has been attained.
Flush by Interval - enforces a file flush after a defined time interval (in seconds).

It's worth noting that the interval flush is a continuous process that acts as a fail-safe mechanism, ensuring that files are periodically flushed, even if the other flush options are not configured or haven't reached their thresholds.

Consider a scenario where the flush size is set to 10MB, and only 9.8MB of data has been written to the file, with no new Kafka messages arriving for an extended period of 6 hours. To prevent undue delays, the interval flush guarantees that the file is flushed after the specified time interval has elapsed. This ensures the timely management of data even in situations where other flush conditions are not met.

The flush options are configured using the batchCount, batchSize and `timeInterval properties. The settings are optional and if not specified the defaults are:

Field

Default

Configuration Examples

Some configuration examples follow on how to apply this connector to different message types.

These include converters, which are required to instruct Kafka Connect on how to read the source content.

Static string template

In this case the converters are irrelevant as we are not using the message content to populate our message template.

Dynamic string template

The HTTP request body contains the value of the message, which is retained as a string value via the StringConverter.

Dynamic string template containing json message fields

Specific fields from the JSON message are substituted into the HTTP request body alongside some static content.

Dynamic string template containing whole json message

The entirety of the message value is substituted into a placeholder in the message body. The message is treated as a string via the StringConverter.

Dynamic string template containing avro message fields

Fields from the AVRO message are substituted into the message body in the following example:

Error/Success Reporter

Starting from version 8.1 as pilot release we give our customers ability to use functionality called Reporter which (if enabled) writes Success and Error processing reports to specified Kafka topic. Reports don't have key and you can find details about status in the message headers and value.

In order to enable this functionality we have to enable one (or both if we want full reporting) of the properties below:

These settings configure the Kafka producer for success and error reports. Full configuration options are available in the and sections. Three examples follow:

Plain Error Reporting

This is the most common scenario for on-premises Kafka Clusters used just for monitoring

Error Reporting using SASL

Using SASL provides a secure and standardized method for authenticating connections to an external Kafka cluster. It is especially valuable when connecting to clusters that require secure communication, as it supports mechanisms like SCRAM, GSSAPI (Kerberos), and OAuth, ensuring that only authorized clients can access the cluster. Additionally, SASL can help safeguard credentials during transmission, reducing the risk of unauthorized access.

Error Reporting using SSL

Using SSL ensures secure communication between clients and the Kafka cluster by encrypting data in transit. This prevents unauthorized parties from intercepting or tampering with sensitive information. SSL also supports mutual authentication, allowing both the client and server to verify each other’s identities, which enhances trust and security in the connection.

Options

Configuration parameters

This sink connector supports the following options as part of its configuration:

Field

Type

Required

Values (Default)

SSL Configuration Properties

Error Reporter Properties

The error reporter can also be configured with SSL Properties. See the section . In this case all properties should be prefixed with connect.reporting.error.config to ensure they apply to the error reporter.

Success Reporter Properties

The error reporter can also be configured with SSL Properties. See the section . In this case all properties should be prefixed with connect.reporting.success.config to ensure they apply to the success reporter.

GCP Storage

This page describes the usage of the Stream Reactor GCP Storage Source Connector.

Connector Class

Example

For more examples see the .

KCQL Support

You can specify multiple KCQL statements separated by ; to have the connector sink into multiple topics.

However, you can not route the same source to different topics, for this use a separate connector instance.

The connector uses a SQL-like syntax to configure the connector behaviour. The full KCQL syntax is:

Please note that you can employ escaping within KCQL for the INSERT INTO, SELECT * FROM, and PARTITIONBY clauses when necessary. For example, if you need to use a topic name that contains a hyphen, you can escape it as follows:

Source Bucket & Path

The GCP Storage source location is defined within the FROM clause. The connector will read all objects from the given location considering the data partitioning and ordering options. Each data partition will be read by a single connector task.

The FROM clause format is:

If your data in GCS was not written by the Lenses GCS sink set to traverse a folder hierarchy in a bucket and load based on the last modified timestamp of the objects in the bucket. If LastModified sorting is used, ensure objects do not arrive late, or use a post-processing step to handle them.

connect.gcpstorage.source.partition.extractor.regex=none

connect.gcpstorage.source.ordering.type=LastModified

Target Bucket & Path

The target Kafka topic is specified via the INSERT INTO clause. The connector will write all the records to the given topic:

GCP Storage object formats

The connector supports a range of storage formats, each with its own distinct functionality:

JSON: The connector will read objects containing JSON content, each line representing a distinct record.
Avro: The connector will read Avro-stored messages from GCP Storage and translate them into Kafka’s native format.
Parquet: The connector will read Parquet-stored messages from GCP Storage and translate them into Kafka’s native format.

Use the STOREAS clause to configure the storage format. The following options are available:

Text Processing

When using Text storage, the connector provides additional configuration options to finely control how text content is processed.

Regex

In Regex mode, the connector applies a regular expression pattern, and only when a line matches the pattern is it considered a record. For example, to include only lines that start with a number, you can use the following configuration:

Start-End line

In Start-End Line mode, the connector reads text content between specified start and end lines, inclusive. This mode is useful when you need to extract records that fall within defined boundaries. For instance, to read records where the first line is ‘SSM’ and the last line is an empty line (’’), you can configure it as follows:

To trim the start and end lines, set the read.text.trim property to true:

Start-End tag

In Start-End Tag mode, the connector reads text content between specified start and end tags, inclusive. This mode is particularly useful when a single line of text in S3 corresponds to multiple output Kafka messages. For example, to read XML records enclosed between ‘’ and ‘’, configure it as follows:

Storage output matrix

Depending on the storage format of Kafka topics’ messages, the need for replication to a different cluster, and the specific data analysis requirements, there exists a guideline on how to effectively utilize converters for both sink and source operations. This guidance aims to optimize performance and minimize unnecessary CPU and memory usage.

S3 Storage Format

Kafka Output Format

Restore or replicate cluster

Analytics

Sink Converter

Source Converter

Projections

Currently, the connector does not offer support for SQL projection; consequently, anything other than a SELECT * query is disregarded. The connector will faithfully write all the record fields to Kafka exactly as they are.

Ordering

[^3]s to ensure precise ordering, leveraging optimizations offered by the GCS API, guaranteeing the accurate sequence of objects.

When using the GCS source alongside the GCS sink, the connector can adopt the same ordering method, ensuring data processing follows the correct chronological order. However, there are scenarios where GCS data is generated by applications that do not maintain lexical object name order.

In such cases, to process objects in the correct sequence, the source needs to list all objects in the bucket and sort them based on their last modified timestamp. To enable this behavior, set the connect.gcpstorage.source.ordering.type to LastModified. This ensures that the source correctly arranges and processes the data based on the timestamps of the objects.

If using LastModified sorting, ensure objects do not arrive late, or use a post-processing step to handle them.

Throttling

To limit the number of object names the source reads from GCS in a single poll. The default value, if not specified, is 1000:

To limit the number of result rows returned from the source in a single poll operation, you can use the LIMIT clause. The default value, if not specified, is 10000.

Object Extension Filtering

The GCP Storage Source Connector allows you to filter the objects to be processed based on their extensions. This is controlled by two properties: connect.gcpstorage.source.extension.excludes and connect.gcpstorage.source.extension.includes.

Excluding Object Extensions

The connect.gcpstorage.source.extension.excludes property is a comma-separated list of object extensions to exclude from the source object search. If this property is not configured, all objects are considered. For example, to exclude .txt and .csv objects, you would set this property as follows:

Including Object Extensions

The connect.gcpstorage.source.extension.includes property is a comma-separated list of object extensions to include in the source object search. If this property is not configured, all objects are considered. For example, to include only .json and .xml objects, you would set this property as follows:

Note: If both connect.gcpstorage.source.extension.excludes and connect.gcpstorage.source.extension.includes are set, the connector first applies the exclusion filter and then the inclusion filter.

Post-Processing Options

Post-processing options offer flexibility in managing how objects are handled after they have been processed. By configuring these options, users can automate tasks such as deleting objects to save storage space or moving objects to an archive for compliance and data retention purposes. These features are crucial for efficient data lifecycle management, particularly in environments where storage considerations or regulatory requirements dictate the need for systematic handling of processed data.

Use Cases for Post-Processing Options

Deleting objects After Processing
For scenarios where freeing up storage is critical and reprocessing is not necessary, configure the connector to delete objects after they are processed. This option is particularly useful in environments with limited storage capacity or where processed data is redundantly stored elsewhere.
Example:
Result: objects are permanently removed from the S3 bucket after processing, effectively reducing storage usage and preventing reprocessing.
Moving objects to an Archive Bucket

Properties

The PROPERTIES clause is optional and adds a layer of configuration to the connector. It enhances versatility by permitting the application of multiple configurations (delimited by ‘,’). The following properties are supported:

Name

Description

Type

Available Values

Authentication

The connector offers two distinct authentication modes:

Default: This mode relies on the default GCP authentication chain, simplifying the authentication process.
File: This mode uses a local (to the connect worker) path for a file containing GCP authentication credentials.
Credentials: In this mode, explicit configuration of a GCP Credentials string is required for authentication.

The simplest example to configure in the connector is the “Default” mode, as this requires no other configuration.

When selecting the “Credentials” mode, it is essential to provide the necessary credentials. Alternatively, if you prefer not to configure these properties explicitly, the connector will follow the credentials retrieval order as described .

Here’s an example configuration for the “Credentials” mode:

And here is an example configuration using the “File” mode:

Remember when using file mode the file will need to exist on every worker node in your Kafka connect cluster and be readable by the Kafka Connect process.

For enhanced security and flexibility when using the “Credentials” mode, it is highly advisable to utilize Connect Secret Providers. This approach ensures robust security practices while handling access credentials.

Backup and Restore

When used in tandem with the GCP Storage Sink Connector, the GCP Storage Source Connector becomes a powerful tool for restoring Kafka topics from GCP Storage. To enable this behavior, you should set store.envelope to true. This configuration ensures that the source expects the following data structure in GCP Storage:

When the messages are sent to Kafka, the GCP Storage Source Connector ensures that it correctly maps the key, value, headers, and metadata fields (including timestamp and partition) to their corresponding Kafka message fields. Please note that the envelope functionality can only be used with data stored in GCP Storage as Avro, JSON, or Parquet formats.

Partition Extraction

When the envelope feature is not in use, and data restoration is required, the responsibility falls on the connector to establish the original topic partition value. To ensure that the source correctly conveys the original partitions back to Kafka Connect during reads from the source, a partition extractor can be configured to extract this information from the GCP Storage object key.

To configure the partition extractor, you can utilize the connect.gcpstorage.source.partition.extractor.type property, which supports two options:

hierarchical: This option aligns with the default format used by the sink, topic/partition/offset.json.
regex: When selected, you can provide a custom regular expression to extract the partition information. Additionally, when using the regex option, you must also set the connect.gcpstorage.source.partition.extractor.regex property. It’s important to note that only one lookup group is expected. For an example of a regular expression pattern, please refer to the pattern used for hierarchical, which is:

Option Reference

Name

Description

Type

Available Values

Default Value

Cassandra

This page describes the usage of the Stream Reactor Cassandra Sink Connector bundled in kafka-connect-cassandra-sink artifact.

Stream Reactor Cassandra Sink Connector is designed to move data from Apache Kafka to Apache Cassandra . It supports Apache Cssandra 3.0 and later, DataStax Enterprise 4.7 and later, and DataStax Astra databases.

Connector Class

The release artifact is called kafka-connect-casandra-sink.

Features

Database authentication : User-password, LDAP, Kerberos
Input formats: The connector supports Avro, Json Schema, Protobuf, Json (schemaless) and primitive data formats. For Schema-Registry based formats, like AVRO, Protobuf, Json-schema, it needs the the Scheam Registry settings.
At least once semantics : The Connect framework stores record offsets in Kafka and resumes from the last committed offset on restart. This ensures reliable delivery but may occasionally result in duplicate record processing during failures

Example

For more examples see the .

Connection

The connector requires a connection to the database. To enable it, specify the following configuration entries. See Configuration Reference for details.

Security

SSL connection

When the database cluster has enabled client encryption, configure the SSL Keys and certificates:

User-Password or LDAP authentication

When the database is configured with user-password of LDAP authentication, configure the following:

Kerberos authentication

When the database cluster has Kerberos authentication enabled set the following settings:

KCQL support

KCQL (Kafka Connect Query Language) is a SQL-like syntax that provides a declarative way to configure connector behavior. It serves as the primary interface for defining how data flows from source topics to target Cassandra tables.

Purpose and Functionality

KCQL statements define three core aspects of data pipeline configuration:

Data Source and Target: Specifies which Kafka topic serves as the data source and which Cassandra table receives the data.

Field Mapping: Controls how fields from the source topic map to columns in the target table, including transformations and selective field inclusion.

Operational Parameters: Configures advanced settings such as consistency levels, delete operations, time-to-live (TTL) settings, and timestamp handling.

Key Capabilities

Flexible Field Mapping: Map source fields to target columns with optional transformations
Consistency Control: Set read and write consistency levels for Cassandra operations
Delete Operations: Enable or disable delete functionality based on message content

Syntax Template

The basic KCQL statement follows this structure:

INSERT INTO: Specifies the target Cassandra keyspace and table where data will be written.

SELECT: Defines field projection and mapping from the source.

FROM: Identifies the source Kafka topic containing the data to be processed.

PROPERTIES: Optional clause for configuring connector behavior such as consistency levels, TTL settings, and operational parameters.

Examples

Table mapping

Kafka to Database Table Mapping

The connector enables mapping fields from Kafka records to database table columns. A single connector can handle multiple topics, where each topic can map to one or more tables. Kafka messages consist of a Key, Value, and Header.

Generic Mapping Syntax

Use the following syntax to map Kafka message fields to database columns:

Prefix the field with:
- _value: For fields from the Kafka message's Value component.
- _key: For fields from the Kafka message's Key component.

Note: If no prefix is provided, _value is assumed, mapping to the Kafka record's Value component.

The record Key and Value can be mapped to specific columns like this, considering a table with row_key and content columns:

To map database columns with whitespace to KCQL projections, follow this format:

Fan-out

You can map a topic's data to multiple database tables using multiple KCQL statements in the same connect.cassandra.kcql configuration. For example:

User defined types

The connector can map complex types from a Kafka record payload into a user-defined type (UDT) column in the database. The incoming data field names must match the database field names.

Consider the following Kafka record format:

key

value

and database definition:

The mapping configuration should be:

Using now() function

You can leverage the now() which returns TIMEUUID function in the mapping. Here is how to use it in KCQL projection:

Write Timestamp

To specify an internal write-time timestamp from the database, choose a numeric field from the Kafka record payload. Use the following mapping syntax:

Examples:

Row-Level TTL

To define the time-to-live (TTL) for a database record, you can optionally map a field from the Kafka record payload. Use the following query:

Examples:

These examples demonstrate specifying TTL units such as 'MINUTES', 'HOURS', or 'DAYS', in addition to the default 'SECONDS'.

Converting date and time for a topic

To configure data and time conversion properties for a topic, use the following SQL command:

Codec Parameter Descriptions:

Parameter

Description

Default

CQL Queries

Advanced CQL Query Configuration for Kafka Records

When a new Kafka record arrives, you have the option to run a custom CQL query. This feature is designed for advanced use cases; typically, the standard Kafka mapping suffices without needing a query. If you specify a query in the topic-to-table mapping, it will take precedence over the default action.

Important: You must include the bound variables used in the query within the mapping column.

Example

Extra settings

Parameter

Description

Default

Java Driver settings

Using the connector configuration, prefix with

any specific setting you would want to set for the Cassandra client. Refer to the for more information.

Configuration Reference

Setting

Description

Default Value

Type

Error policies

The connector supports .

Azure Data Lake Gen2

This page describes the usage of the Stream Reactor Azure Datalake Gen 2 Sink Connector.

This Kafka Connect sink connector facilitates the seamless transfer of records from Kafka to Azure Data Lake Buckets. It offers robust support for various data formats, including AVRO, Parquet, JSON, CSV, and Text, making it a versatile choice for data storage. Additionally, it ensures the reliability of data transfer with built-in support for exactly-once semantics.

Connector Class

Example

For more examples see the .

KCQL Support

You can specify multiple KCQL statements separated by ; to have a connector sink multiple topics. The connector properties topics or topics.regex are required to be set to a value that matches the KCQL statements.

The connector uses KCQL to map topics to Datalake buckets and paths. The full KCQL syntax is:

Please note that you can employ escaping within KCQL for the INSERT INTO, SELECT * FROM, and PARTITIONBY clauses when necessary. For example, an incoming Kafka message stored as JSON can use fields containing .:

In this case, you can use the following KCQL statement:

Target Bucket and Path

The target bucket and path are specified in the INSERT INTO clause. The path is optional and if not specified, the connector will write to the root of the bucket and append the topic name to the path.

Here are a few examples:

SQL Projection

Currently, the connector does not offer support for SQL projection; consequently, anything other than a SELECT * query is disregarded. The connector will faithfully write all fields from Kafka exactly as they are.

Source Topic

To avoid runtime errors, make sure the topics or topics.regex setting matches your KCQL statements. If the connector receives data for a topic without matching KCQL, it will throw an error. When using a regex to select topics, follow this KCQL pattern:

In this case the topic name will be appended to the $target destination.

KCQL Properties

The PROPERTIES clause is optional and adds a layer of configuration to the connector. It enhances versatility by permitting the application of multiple configurations (delimited by ‘,’). The following properties are supported:

Name

Description

Type

Available Values

Default Value

The sink connector optimizes performance by padding the output files, a practice that proves beneficial when using the Datalake Source connector to restore data. This file padding ensures that files are ordered lexicographically, allowing the Datalake Source connector to skip the need for reading, sorting, and processing all files, thereby enhancing efficiency.

Partitioning & File names

The object key serves as the filename used to store data in Datalake. There are two options for configuring the object key:

Default: The object key is automatically generated by the connector and follows the Kafka topic-partition structure. The format is $container/[$prefix]/$topic/$partition/offset.extension. The extension is determined by the chosen storage format.
Custom: The object key is driven by the PARTITIONBY clause. The format is either $container/[$prefix]/$topic/customKey1=customValue1/customKey2=customValue2/topic(partition_offset).extension (naming style mimicking Hive-like data partitioning) or $container/[$prefix]/customValue/topic(partition_offset).ext. The extension is determined by the selected storage format.

The Connector automatically adds the topic name to the partition. There is no need to add it to the partition clause. If you want to explicitly add the topic or partition you can do so by using _topic and _partition.

The partition clause works on Header, Key and Values fields of the Kafka message.

Custom keys and values can be extracted from the Kafka message key, message value, or message headers, as long as the headers are of types that can be converted to strings. There is no fixed limit to the number of elements that can form the object key, but you should be aware of Azure Datalake key length restrictions.

To extract fields from the message values, simply use the field names in the PARTITIONBY clause. For example:

However, note that the message fields must be of primitive types (e.g., string, int, long) to be used for partitioning.

You can also use the entire message key as long as it can be coerced into a primitive type:

In cases where the Kafka message Key is not a primitive but a complex object, you can use individual fields within the message Key to create the Datalake object key name:

Kafka message headers can also be used in the Datalake object key definition, provided the header values are of primitive types easily convertible to strings:

Customizing the object key can leverage various components of the Kafka message. For example:

This flexibility allows you to tailor the object key to your specific needs, extracting meaningful information from Kafka messages to structure Datalake object keys effectively.

To enable Athena-like partitioning, use the following syn

Rolling Windows

Storing data in Azure Datalake and partitioning it by time is a common practice in data management. For instance, you may want to organize your Datalake data in hourly intervals. This partitioning can be seamlessly achieved using the PARTITIONBY clause in combination with specifying the relevant time field. However, it’s worth noting that the time field typically doesn’t adjust automatically.

To address this, we offer a Kafka Connect Single Message Transformer (SMT) designed to streamline this process. You can find the transformer plugin and documentation .

Let’s consider an example where you need the object key to include the wallclock time (the time when the message was processed) and create an hourly window based on a field called timestamp. Here’s the connector configuration to achieve this:

In this example, the incoming Kafka message’s Value content includes a field called timestamp, represented as a long value indicating the epoch time in milliseconds. The TimestampConverter SMT will expertly convert this into a string value according to the format specified in the format.to.pattern property. Additionally, the insertWallclock SMT will incorporate the current wallclock time in the format you specify in the format property.

The PARTITIONBY clause then leverages both the timestamp field and the wallclock header to craft the object key, providing you with precise control over data partitioning.

Data Storage Format

While the STOREAS clause is optional, it plays a pivotal role in determining the storage format within Azure Datalake. It’s crucial to understand that this format is entirely independent of the data format stored in Kafka. The connector maintains its neutrality towards the storage format at the topic level and relies on the key.converter and value.converter settings to interpret the data.

Supported storage formats encompass:

AVRO
Parquet
JSON
CSV (including headers)

Opting for BYTES ensures that each record is stored in its own separate file. This feature proves particularly valuable for scenarios involving the storage of images or other binary data in Datalake. For cases where you prefer to consolidate multiple records into a single binary file, AVRO or Parquet are the recommended choices.

By default, the connector exclusively stores the Kafka message value. However, you can expand storage to encompass the entire message, including the key, headers, and metadata, by configuring the store.envelope property as true. This property operates as a boolean switch, with the default value being false. When the envelope is enabled, the data structure follows this format:

Not supported with a custom partition strategy.

Utilizing the envelope is particularly advantageous in scenarios such as backup and restore or replication, where comprehensive storage of the entire message in Datalake is desired.

Examples

Storing the message Value Avro data as Parquet in Datalake:

The converter also facilitates seamless JSON to AVRO/Parquet conversion, eliminating the need for an additional processing step before the data is stored in Datalake.

Enabling the full message stored as JSON in Datalake:

Enabling the full message stored as AVRO in Datalake:

If the restore (see the Datalake Source documentation) happens on the same cluster, then the most performant way is to use the ByteConverter for both Key and Value and store as AVRO or Parquet:

Flush Options

The connector offers three distinct flush options for data management:

Flush by Count - triggers a file flush after a specified number of records have been written to it.
Flush by Size - initiates a file flush once a predetermined size (in bytes) has been attained.
Flush by Interval - enforces a file flush after a defined time interval (in seconds).

It’s worth noting that the interval flush is a continuous process that acts as a fail-safe mechanism, ensuring that files are periodically flushed, even if the other flush options are not configured or haven’t reached their thresholds.

The flush options are configured using the flush.count, flush.size, and flush.interval KCQL Properties (see section). The settings are optional and if not specified the defaults are:

flush.count = 50_000
flush.size = 500000000 (500MB)
flush.interval = 3600 (1 hour)

A connector instance can simultaneously operate on multiple topic partitions. When one partition triggers a flush, it will initiate a flush operation for all of them, even if the other partitions are not yet ready to flush.

When connect.datalake.latest.schema.optimization.enabled is set to true, it reduces unnecessary data flushes when writing to Avro or Parquet formats. Specifically, it leverages schema compatibility to avoid flushing data when messages with older but backward-compatible schemas are encountered. Consider the following sequence of messages and their associated schemas:

Flushing By Interval

The next flush time is calculated based on the time the previous flush completed (the last modified time of the file written to Data Lake). Therefore, by design, the sink connector’s behaviour will have a slight drift based on the time it takes to flush records and whether records are present or not. If Kafka Connect makes no calls to put records, the logic for flushing won't be executed. This ensures a more consistent number of records per file.

Compression

AVRO and Parquet offer the capability to compress files as they are written. The GCP Storage Sink connector provides advanced users with the flexibility to configure compression options.

Here are the available options for the connect.gcpstorage.compression.codec, along with indications of their support by Avro, Parquet and JSON writers:

Compression

Avro Support

Avro (requires Level)

Parquet Support

JSON

Please note that not all compression libraries are bundled with the Datalake connector. Therefore, you may need to manually add certain libraries to the classpath to ensure they function correctly.

Authentication

The connector offers two distinct authentication modes:

Default: This mode relies on the default Azure authentication chain, simplifying the authentication process.
Connection String: This mode enables simpler configuration by relying on the connection string to authenticate with Azure.
Credentials: In this mode, explicit configuration of Azure Access Key and Secret Key is required for authentication.

When selecting the “Credentials” mode, it is essential to provide the necessary access key and secret key properties. Alternatively, if you prefer not to configure these properties explicitly, the connector will follow the credentials retrieval order as described here.

Here’s an example configuration for the “Credentials” mode:

And here is an example configuration using the “Connection String” mode:

For enhanced security and flexibility when using either the “Credentials” or “Connection String” modes, it is highly advisable to utilize Connect Secret Providers.

Error policies

The connector supports .

Indexes Directory

The connector uses the concept of index files that it writes to in order to store information about the latest offsets for Kafka topics and partitions as they are being processed. This allows the connector to quickly resume from the correct position when restarting and provides flexibility in naming the index files.

By default, the root directory for these index files is named .indexes for all connectors. However, each connector will create and store its index files within its own subdirectory inside this .indexes directory.

You can configure the root directory for these index files using the property connect.datalake.indexes.name. This property specifies the path from the root of the data lake filesystem. Note that even if you configure this property, the connector will still create a subdirectory within the specified root directory.

Examples

Option Reference

Name

Description

Type

Available Values

Default Value

Stream Reactor

This page contains the release notes for the Stream Reactor.

11.6.0

Bug Fix: DataLake Sinks (S3, GCS, Azure Datalake Gen2)

When String and Enum are used as a field type, connectors may fail to transfer data to the target. The enhancement adds better Avro schema support, preserving package information and Enums during conversions between Avro and Connect schemas.

11.5.0

Bug Fix: DataLake Sinks (S3, GCS, Azure Datalake Gen2)

When v3 records arrived before v2 records, an ArrayIndexOutOfBoundsException occurred. This happened because ParquetWriter or AvroWriter attempted to write GenericRecords with mismatched Avro schema objects.

11.4.0

Improvement: DataLake Sinks (S3, GCS, Azure Datalake Gen2)

With at-least-once semantics enabled, the sink connector uploads the file directly to the destination. This contrasts with the three operations needed for exactly-once delivery. The result is improved latency and reduced cloud costs.

Fix: GCP Pub/Sub Source

Previously, each subscription was limited to a single task, restricting scalability. This update now allows subscriptions to be distributed across all tasks in the connector, enhancing capacity and performance.

11.3.0

Fix: S3 path vs virtual hosted URL style

This release addresses a semantic inconsistency in how S3 access styles were configured, specifically affecting users of S3-compatible storage solutions (e.g., MinIO, s3proxy) or specific AWS regions requiring path-style access.

The Issue

Previously, the configuration property connect.s3.vhost.bucket was mapped directly to the AWS SDK’s pathStyleAccessEnabled() method. However, these two concepts are semantically opposite:

AWS SDK pathStyleAccessEnabled(true): Forces Path-Style access (e.g., http://endpoint/bucket).
Connector Property vhost.bucket=true: Implies Virtual-Hosted access (e.g., http://bucket.endpoint).

Because the connector passed the boolean value directly without inversion, users were forced to use a counter-intuitive configuration (setting vhost.bucket=true to achieve path-style access) to support storage systems that do not support virtual hosting.

The Fix

We have introduced a new, unambiguous configuration property: connect.s3.path.style.access.

true: Enables path-style access (matches AWS SDK semantics).
false: Enables virtual-hosted style access.

Resolution & Backward Compatibility

To maintain backward compatibility for existing deployments, the logic now prioritizes the new property while preserving the legacy behavior for the old property.

Configuration Priority Logic:

If connect.s3.path.style.access is set:
- The connector uses this value.
- connect.s3.vhost.bucket is ignored.

Migration Guide

DataLake sources late handling of data

In scenarios where users cannot ensure files are uploaded to cloud storage sequentially or by timestamp, late-arriving files—those older than the current watermark—are often skipped and never processed. Previously, the connector would only process files newer than this watermark. This update introduces the post.process.action.watermark.process.late.arrival KCQL property, enabling files uploaded out of order to be processed when the Move post-process action is used.

This approach periodically scans the data lake for files with timestamps older than the connector's watermark. It then copies them to themselves to update their last modified timestamp, which may incur additional costs. Users are encouraged to ensure their systems can produce files in sequence to avoid these costs and potential order discrepancies, as out-of-order processing affects record sequencing.

The scan interval is defined at the connector level:

For AWS S3 is: connect.s3.source.late.arrival.interval
For GCS is: connect.gcpstorage.source.late.arrival.interval

DataLake sink control the object key namer

Custom File Namer Factory

A custom FileNamerFactory can be created with an optional configuration parameter. The FileNamerFactory includes a method to generate a FileNamer based on existing system parameters for file naming.

The KCQL properties for data lake sinks now support setting store.file.namer to a specific class. Users must currently provide their own class and implementation, with potential enhancements anticipated in future updates.

11.2.0

Azure Data Lake Gen2 Sink Update

The patch release addresses several issues discovered in the Azure Data Lake Storage (ADLS) Gen2 sink:

412 ConditionNotMet Errors: Occurred during lock file updates.
Empty File Conflicts: Resulted from recursive directory marker creation.
Unexpected Whitespace or EOF Errors: Caused by an incorrect state, leading to the lock file's eTag being returned as the lock file's content.

11.1.0 DataLakes Sinks

This release resolves a gap in the S3, GCS, and Azure sinks when latest schema optimization is enabled (via connect.***.latest.schema.optimization.enabled). Previously, connectors would fail if an Avro field changed from an Enum to a Union of an Enum and String. This update fixes that problem.

11.0.0

This release upgrades the connectors to Apache Kafka 4.1. All connectors have been tested and verified against Kafka versions 4.0 and 4.1.

Compatibility Notice

Important: Kafka 3.x versions are no longer supported or tested with this release. While the connectors may function with Kafka 3.x environments, they are not recommended for production use without thorough testing and validation in your specific environment.

Organizations currently running Kafka 3.x should plan to upgrade their Kafka infrastructure to version 4.0 or later before deploying these connectors in production.

Connector Retirement

Effective with version 11.0.0, the following sink connectors have been retired and are no longer available:

Redis Sink Connector
InfluxDB Sink Connector

These connectors will not receive further updates, bug fixes, or security patches. Organizations currently using these connectors should evaluate alternative solutions and plan migration strategies accordingly.

For questions regarding migration paths or alternative connector options, please consult the product documentation or contact support.

10.0.3

Datalakes sinks (AWS S3, GCS and Azure DataLake Gen2)

The sink addresses a gap by committing the message offset as-is, rather than incrementing it by one. This behavior causes Kafka's underlying consumer to appear as if it's always one message behind each partition. This isn't an issue under exactly-once semantics, as the state store in the data lake ensures message processing integrity, ignoring any already processed messages. Under at-least-once semantics, replaying a message is considered acceptable.

10.0.2

Azure DataLake Gen 2

This patch release addresses two issues on the Azure DataLake sink

Fixed errors caused by malformed HTTP headers in ADLS Gen2 requests, such as:
Fixed failures when writing to nested directories in ADLS Gen2 with Hierarchical Namespace (HNS) enabled.

Thank you to and for their fixes.

10.0.1

DataLakes

This patch addresses several critical bugs and regressions introduced in version 9.0.0, affecting the Datalake connectors for S3, GCS, and Azure.

Connector Restart Logic: Fixed an issue that caused an error on restart if pending operations from a previous session had not completed.
S3 Connector: Corrected a bug where delete object requests incorrectly sent an empty versionId. This resolves API exceptions when using S3-compatible storage solutions, such as Pure Storage.
Addressed a regression where the connector name was omitted from the lock path for exactly-once processing. This fix prevents conflicts when multiple connectors read from the same Kafka topics and write to the same destination bucket.

10.0.0

New Apache Cassandra sink

This new connector is compatible with the any Apache Cassandra compatible databases from version 3+. The previous connector is deprecated and will be removed in a future release

Azure CosmosDB sink

Version 10.0.0 introduces breaking changes: the connector has been renamed from DocumentDB, uses the official CosmosDB SDK, and supports new bulk and key strategies.

The CosmosDB Sink Connector is a Kafka Connect sink connector designed to write data from Kafka topics into Azure CosmosDB. It was formerly known as the DocumentDB Sink Connector and has been fully renamed and refactored with the following key changes:

Renamed from DocumentDB to CosmosDB across all code and documentation.
Updated package names to reflect the new CosmosDB naming.
Replaced the legacy DocumentDB client with the official CosmosDB Java SDK.

New Features and Improvements

In non-envelope mode, DataLakes connector sinks now skip tombstone records, preventing connector failures.
HTTP Sink: Enable Import of All Headers in the Message Template

9.0.2

DataLake Sinks (S3, GCP Storage, Azure Data Lake)

The previous commit that addressed the Exactly Once semantics fix inadvertently omitted support for the connect.xxxxxxx.indexes.name configuration. As a result, custom index filenames were not being applied as expected.

This commit restores that functionality by ensuring the connect.xxxxxxx.indexes.name setting is properly respected during index file generation.

9.0.1

Google BigQuery Sink

The latest release of Stream Reactor includes a fork of Apache 2's Confluent BigQuery, originally sourced from WePay.

9.0.0

All Modules

Dependency upgrades.
Project now builds for Kafka 3.9.1.
Exclude Kafka dependencies and certain other things (eg log frameworks) from the final jar.

DataLake Sinks (S3, GCP Storage, Azure Data Lake)

Improved Exactly Once Semantics by introducing a per-task lock mechanism to prevent duplicate writes when Kafka Connect inadvertently runs multiple tasks for the same topic-partition.
- Each task now creates a lock file per topic-partition (e.g., lockfile_topic_partition.yml) upon startup.
- The lock file is used to ensure only one task (the most recent) can write to the target bucket, based on object eTag validation.

MQTT Connector

Refactored MqttWriter to improve payload handling and support dynamic topics, with JSON conversion for structured data and removal of legacy converters and projections (#232).
Leading Slashes Fixes:
- Leading slash from MQTT topic should be removed when converting to Kafka topic target.

8.1.35

Bug Fix: DataLake Sinks (S3, GCS, Azure Datalake Gen2)

8.1.34

Bug Fix: DataLake Sinks (S3, GCS, Azure Datalake Gen2)

8.1.33

DataLake Sinks (S3, GCP Storage, Azure Data Lake)

This update introduces an optimization that reduces unnecessary data flushes when writing to Avro or Parquet formats. Specifically, it leverages schema compatibility to avoid flushing data when messages with older but backward-compatible schemas are encountered.

How It Works

When the input message schema changes in a backward-compatible way, the sink tracks the latest schema and continues using it for serialization. This means that incoming records with an older schema won't automatically trigger a flush, as long as the older schema is compatible with the current one.

Example Flush Behavior

Consider the following sequence of messages and their associated schemas:

How to enable it

Set the following config to true:

S3: connect.s3.latest.schema.optimization.enabled
GCP: connect.gcpstorage.latest.schema.optimization.enabled
Azure DataLake: connect.datalake.latest.schema.optimization.enabled

8.1.32

GCP Storage Sink

Reduces log entries, by moving the log entry confirming the object upload from INFO to DEBUG

8.1.31

DataLake Sinks

This release introduces improvements to Avro error handling, providing better diagnostics and insights into failures during data writing.

8.1.30

DataLake Sinks (S3, GCP Storage, Azure Data Lake)

Resolved dependency issues causing sink failures with Confluent 7.6.0. Confluent 7.6.0 introduced new Schema Registry rule modules that were force-loaded, even when unused, leading to the following error:
This update ensures compatibility by adjusting dependencies, preventing unexpected failures in affected sinks.

HTTP Sink

Adjusting the following log line from INFO level to TRACE level

8.1.29

DataLake Sinks (S3, GCP Storage, Azure Data Lake)

Support has been added for writing to the same bucket from two different connectors located in different Kafka clusters, with both reading from the same topic name. To differentiate the object keys generated and prevent data loss, a new KCQL syntax has been introduced:\

HTTP Sink

This feature offers fixed interval retries in addition to the Exponential option. To enable it, use the following configuration settings

A bug affecting the P99, P90, and P50 metrics has been resolved. Previously, failed HTTP requests were not included in the metrics calculation.

8.1.28

DataLake Sinks (S3, GCP Storage, Azure Data Lake)

Enable Skipping of Null Records

Added the following configuration property to prevent possible NullPointerException situations in the S3 Sink, Azure Datalake Sink, and GCP Storage Sink connectors:

S3 Sink: connect.s3.skip.null.values
GCP Storage Sink: connect.gcpstorage.skip.null.values
Azure Datalake Sink: connect.datalake.skip.null.values

When set to true, the sinks will skip null value records instead of processing them. Defaults to false.

If you expect null or tombstone records and are not using envelope mode, enabling this setting is recommended to avoid errors.

HTTP Sink

Fix MBean Registration Issue

Fixed an issue where MBean registration failed when multiple tasks were used, causing exceptions due to duplicate instances. The MBean name now includes the task ID to ensure uniqueness, and MBeans are properly unregistered when tasks stop.

8.1.27

Reverted: Change to LastModified Ordering with Post-Process Actions (introduced in 8.1.23)

We have reverted the change that avoided filtering to the latest result when ordering by LastModified with a post-process action. The original change aimed to prevent inconsistencies due to varying LastModified timestamps.

However, this introduced a potential issue due to Kafka Connect’s commitRecord method being called asynchronously, meaning files may not be cleaned up before the next record is read. This could result in records being processed multiple times. The previous behaviour has been restored to ensure correctness.

8.1.26

Azure Service Bus source

Source watermark is not stored anymore since it is not used when the task restarts

GCP PubSub source

A fix was made to enable attaching the headers to the resulting Kafka message.

8.1.25

All Modules

Align Confluent lib dependency with Kafka 3.8.

AWS S3 Source

Customers have noticed that when the source deletes files on S3, then the entire path is deleted. Whilst this is actually due to the way S3 works, we can actually do something about this in the connector.

This adds a new boolean KCQL property to the S3 source: post.process.action.retain.dirs (default value: false)

If this is set to true, then upon moving/deleting files within the source post-processing, then first a zero-byte object will be created to ensure that the path will still be represented on S3.

8.1.24

HTTP

JMX Metrics Support

The HTTP Sink now publishes counts for success, request average among other metrics to JMX.

Extractor Fix

A bug was reported that referencing the entire record value yields an exception when the payload is a complex type. In this case a HashMap.

The changes allow the return of the value when the extractor has no field to extract (ie. extract the full value)

DataLake Sinks (S3, GCP Storage, Azure Data Lake) Optimised Schema Change

This PR revamps the mechanisms governing schema change rollover, as they can lead to errors. New functionality is introduced to ensure improved compatibility.

Removed Hadoop Shaded Dependency

The hadoop-shaded-protobuf dependency has been removed as it was surplus to requirements and pulling in an old version of protobuf-java which introduced vulnerabilities to the project.

Removed Properties for Schema Change Rollover:

The properties $connectorPrefix.schema.change.rollover have been removed for the following connectors: connect.s3, connect.datalake, and connect.gcpstorage. This change eliminates potential errors and simplifies the schema change handling logic.

New Property for Schema Change Detection:

The property $connectorPrefix.schema.change.detector has been introduced for the following connectors: connect.s3, connect.datalake, and connect.gcpstorage.

This property allows users to configure the schema change detection behavior with the following possible values: • default: Schemas are compared using object equality. • version: Schemas are compared by their version field. • compatibility: A more advanced mode that ensures schemas are compatible using Avro compatibility features.

Packaging Changes

The META-INF/maven directory is now built by the assembly process. This will ensure it reflects what is in the jar and that this directory is not just built from our component jars. This mirrors an approach we took for the secret-provider.

8.1.23

DataLakes (S3, GCP) source fixes

Polling Backoff

The connector incurs high costs when there is no data available in the buckets because it continuously polls the data lake in a tight loop, as controlled by Kafka Connect.

From this version by default a backoff queue is used, introducing a standard method for backing off calls to the underlying cloud platform.

Avoid filtering by lastSeenFile where a post process action is configured

When ordering by LastModified and a post-process action is configured, avoid filtering to the latest result.

This change avoids bugs caused by inconsistent LastModified dates used for sorting. If LastModified sorting is used, ensure objects do not arrive late, or use a post-processing step to handle them.

Add a flag to populate kafka headers with the watermark partition/offset

This adds a connector property for GCP Storage and S3 Sources: connect.s3.source.write.watermark.header connect.gcpstorage.source.write.watermark.header

If set to true then the headers in the source record produced will include details of the source and line number of the file.

If set to false (the default) then the headers won't be set.

Currently this does not apply when using the envelope mode.

8.1.22

DataLakes (S3, GCP) source fixes

This release addresses two critical issues:

Corrupted connector state when DELETE/MOVE is used: The connector is designed to store the last processed document and its location within its state for every message sent to Kafka. This mechanism ensures that the connector can resume processing from the correct point in case of a restart. However, when the connector is configured with a post-operation to move or delete processed objects within the data lake, it stores the last processed object in its state. If the connector restarts and the referenced object has been moved or deleted externally, the state points to a non-existent object, causing the connector to fail. The current workaround requires manually cleaning the state and restarting the connector, which is inefficient and error-prone.
Incorrect Handling of Move Location Prefixes: When configuring the move location within the data lake, if the prefix ends with a forward slash (/), it results in malformed keys like a//b. Such incorrect paths can break compatibility with query engines like Athena, which may not handle double slashes properly.

8.1.21

Azure Service Bus source

Performance improvements in the source to handle a higher throughput. The code now leverages prefetch count, and disables the auto complete. The following connector configs were added

connect.servicebus.source.prefetch.count The number of messages to prefetch from ServiceBus
connect.servicebus.source.complete.retries.max The maximum number of retries to attempt while completing a message
connect.servicebus.source.complete.retries.min.backoff.ms

8.1.20

Datalakes extra logging

A NullPointerException is thrown and the sinks lose the stacktrace. It is a patch to enhance the logging

8.1.19

Prevent ElasticSearch from Skipping Records After Tombstone

Addresses a critical bug in ElasticSearch versions 6 (ES6) and 7 (ES7) where records following a tombstone are inadvertently skipped during the insertion process. The issue stemmed from an erroneous return statement that halted the processing of subsequent records.

8.1.18

🚀 New Features

Sequential Message Sending for Azure Service Bus

Introduced a new KCQL property: batch.enabled (default: true).
Users can now disable batching to send messages sequentially, addressing specific scenarios with large message sizes (e.g., >1 MB).
Why this matters: Batching improves performance but can fail for large messages. Sequential sending ensures reliability in such cases.

Post-Processing for Datalake Cloud Source Connectors

Added post-processing capabilities for AWS S3 and GCP Storage source connectors ( Azure Datalake Gen 2 support coming soon).
New KCQL properties:
- post.process.action: Defines the action (DELETE or MOVE

Example 2: Move files to an archive bucket:

🛠 Dependency Updates

Updated Azure Service Bus Dependencies

azure-core updated to version 1.54.1.
azure-messaging-servicebus updated to version 7.17.6.

These updates ensure compatibility with the latest Azure SDKs and improve stability and performance.

Upgrade Notes

Review the new KCQL properties and configurations for Azure Service Bus and Datalake connectors.
Ensure compatibility with the updated Azure Service Bus dependencies if you use custom extensions or integrations.

Thank you to all contributors! 🎉

8.1.17

Improves ElasticSearch sinks by allowing the message key fields, or message headers to be used as part of the document primary key. Reference _key[.key_field_path or _header.header_name to set the document primary key
Fixes the data lake sinks error message on flush.interval

8.1.16

Improvements to the HTTP Sink

Queue Limiting: We've set a limit on the queue size per topic to reduce the chances of an Out-of-Memory (OOM) issue. Previously the queue was unbounded and in a scenario where the http calls are slow and the sink gets more records than it clears, it would eventually lead to OOM.
Offering Timeout: The offering to the queue now includes a timeout. If there are records to be offered, but the timeout is exceeded, a retriable exception is thrown. Depending on the connector's retry settings, the operation will be attempted again. This helps avoid situations where the sink gets stuck processing a slow or unresponsive batch.
Duplicate Record Handling: To prevent the same records from being added to the queue multiple times, we've introduced a

8.1.15

HTTP sink ignores null records. Avoids a NPE if custom SMTs send null records to the pipeline

8.1.14

Improvements to the HTTP Sink reporter

8.1.13

8.1.12

Datalake sinks allow an optimisation configuration to avoid rolling the file on schema change. To be used only when the schema changes are backwards compatible

Fixes for the HTTP sink batch.

8.1.11

HTTP sink improvements

8.1.10

Dependency version upgrades.

Azure Service Bus Source

Fixed timestamp of the messages and some of the fields in the payload to correctly use UTC millis.

8.1.9

Azure Service Bus Source

Changed contentType in the message to be nullable (Optional String).

8.1.8

HTTP Sink

Possible exception fix

Store success/failure response codes in http for the reporter.

8.1.7

HTTP Sink

Fixes possible exception in Reporter code.

8.1.6

HTTP Sink

Fixes and overall stability improvements.

8.1.4

Google Cloud Platform Storage

Remove overriding transport options on GCS client.

8.1.3

HTTP Sink

Bugfix: HTTP Kafka Reporter ClientID is not unique.

8.1.2

HTTP Sink & GCS Sink

Improvements for handing HTTP sink null values & fix for NPE on GCS.

8.1.1

Removal of shading for Avro and Confluent jars.

8.1.0

HTTP Sink

Addition of HTTP reporter functionally to route HTTP success and errors to a configurable topic.

8.0.0

HTTP Sink

Breaking Configuration Change

Configuration for the HTTP Sink changes from Json to standard Kafka Connect properties. Please study the documentation if upgrading to 8.0.0.

Removed support for JSON configuration format. Instead please now configure your HTTP Sink connector using kafka connect properties. .
Introduced SSL Configuration. .
Introduced OAuth Support. .

7.4.5

ElasticSearch (6 and 7)

Support for dynamic index names in KCQL.
Configurable tombstone behaviour using KCQL property. behavior.on.null.values
SSL Support using standard Kafka Connect properties.

Http Sink

Adjust defaultUploadSyncPeriod to make connector more performant by default.

Data Lake Sinks (AWS, Azure Datalake and GCP Storage)

Fix for issue causing sink to fail with NullPointerException due to invalid offset.

7.4.4

Azure Datalake & GCP Storage

Dependency version upgrades

Data Lake Sinks (AWS, Azure Datalake and GCP Storage)

Fixes a gap in the avro/parquet storage where enums where converted from Connect enums to string.
Adds support for explicit "no partition" specification to kcql, to enable topics to be written in the bucket and prefix without partitioning the data.
- Syntax Example: INSERT INTO foo SELECT * FROM bar NOPARTITION

7.4.3

All Connectors

Dependency version upgrades

Data Lake Sinks (AWS, Azure Datalake and GCP Storage)

This release introduces a new configuration option for three Kafka Connect Sink Connectors—S3, Data Lake, and GCP Storage—allowing users to disable exactly-once semantics. By default, exactly once is enabled, but with this update, users can choose to disable it, opting instead for Kafka Connect’s native at-least-once offset management.

S3 Sink Connector: connect.s3.exactly.once.enable Data Lake Sink Connector: connect.datalake.exactly.once.enable GCP Storage Sink Connector: connect.gcpstorage.exactly.once.enable

Default Value: true

Indexing is enabled by default to maintain exactly-once semantics. This involves creating an .indexes directory at the root of your storage bucket, with subdirectories dedicated to tracking offsets, ensuring that records are not duplicated.

Users can now disable indexing by setting the relevant property to false. When disabled, the connector will utilise Kafka Connect’s built-in offset management, which provides at-least-once semantics instead of exactly-once.

7.4.2

All Connectors

Dependency version upgrades

AWS S3 Source and GCP Storage Source Connectors

File Extension Filtering

Introduced new properties to allow users to filter the files to be processed based on their extensions.

For AWS S3 Source Connector:

connect.s3.source.extension.excludes: A comma-separated list of file extensions to exclude from the source file search. If this property is not configured, all files are considered.
connect.s3.source.extension.includes: A comma-separated list of file extensions to include in the source file search. If this property is not configured, all files are considered.

For GCP Storage Source Connector:

connect.gcpstorage.source.extension.excludes: A comma-separated list of file extensions to exclude from the source file search. If this property is not configured, all files are considered.
connect.gcpstorage.source.extension.includes: A comma-separated list of file extensions to include in the source file search. If this property is not configured, all files are considered.

These properties provide more control over the files that the AWS S3 Source Connector and GCP Storage Source Connector process, improving efficiency and flexibility.

GCP Storage Source and Sink Connectors

Increases the default HTTP Retry timeout from 250ms total timeout to 3 minutes as default. The default consumer group max.poll.timeout is 5 minutes, so it’s within the boundaries to avoid a group rebalance.

7.4.1

GCP Storage Connector

Adding retry delay multiplier as a configurable parameter (with default value) to Google Cloud Storage Connector. Main changes revolve around RetryConfig class and its translation to gax HTTP client config.

Data Lake Sinks (AWS, Azure Datalake and GCP Storage)

Making indexes directory configurable for both source and sink.

Sinks

Use the below properties to customise the indexes root directory:

connect.datalake.indexes.name
connect.gcpstorage.indexes.name
connect.s3.indexes.name

See connector documentation for more information.

Sources:

Use the below properties to exclude custom directories from being considered by the source.
connect.datalake.source.partition.search.excludes
connect.gcpstorage.source.partition.search.excludes

7.4.0

NEW: Azure Service Bus Sink Connector

Azure Service Bus Sink Connector.

Other Changes

Exception / Either Tweaks
Bump java dependencies (assertJ and azure-core)
Add logging on flushing

Full Changelog:

7.3.2

Data Lake Sinks (AWS, Azure Datalake and GCP Storage)

Support added for writing Date, Time and Timestamp data types to AVRO.

7.3.1

GCP Storage Connector

Invalid Protocol Configuration Fix

7.3.0

NEW: Azure Service Bus Source Connector

Azure Service Bus Source Connector

NEW: GCP Pub/Sub Source Connector

GCP Pub/Sub Source Connector

Data Lake Sinks (AWS, Azure Datalake and GCP Storage)

To back the topics up the KCQL statement is

When * is used the envelope setting is ignored.

This change allows for the * to be taken into account as a default if the given message topic is not found.

HTTP Sink

Bug fix to ensure that, if specified as part of the template, the Content-Type header is correctly populated.

All Connectors:

Update of dependencies.

Upgrading from any version prior to 7.0.0, please see the release and upgrade notes for 7.0.0.

7.2.0

Enhancements

Automated Skip for Archived Objects:

The S3 source now seamlessly bypasses archived objects, including those stored in Glacier and Deep Archive. This enhancement improves efficiency by automatically excluding archived data from processing, avoiding the connector crashing otherwise

Enhanced Key Storage in Envelope Mode:

Changes have been implemented to the stored key when using envelope mode. These modifications lay the groundwork for future functionality, enabling seamless replay of Kafka data stored in data lakes (S3, GCS, Azure Data Lake) from any specified point in time.

Full Changelog:

7.1.0

Source Line-Start-End Functionality Enhancements

We’ve rolled out enhancements to tackle a common challenge faced by users of the S3 source functionality. Previously, when an external producer abruptly terminated a file without marking the end message, data loss occurred.

To address this, we’ve introduced a new feature: a property entry for KCQL to signal the handling of unterminated messages. Meet the latest addition, read.text.last.end.line.missing. When set to true, this property ensures that in-flight data is still recognized as a message even when EOF is reached but the end line marker is missing.

Upgrading from any version prior to 7.0.0, please see the release and upgrade notes for 7.0.0.

7.0.0

Data-lakes Sink Connectors

This release brings substantial enhancements to the data-lakes sink connectors, elevating their functionality and flexibility. The focal point of these changes is the adoption of the new KCQL syntax, designed to improve usability and resolve limitations inherent in the previous syntax.

Key Changes

New KCQL Syntax: The data-lakes sink connectors now embrace the new KCQL syntax, offering users enhanced capabilities while addressing previous syntax constraints.
Data Lakes Sink Partition Name: This update ensures accurate preservation of partition names by avoiding the scraping of characters like \ and /. Consequently, SMTs can provide partition names as expected, leading to reduced configuration overhead and increased conciseness.

KCQL Keywords Replaced

Several keywords have been replaced with entries in the PROPERTIES section for improved clarity and consistency:

WITHPARTITIONER: Replaced by PROPERTIES ('partition.include.keys'=true/false). When WITHPARTITIONER KeysAndValue is set to true, the partition keys are included in the partition path. Otherwise, only the partition values are included.
WITH_FLUSH_SIZE: Replaced by PROPERTIES ('flush.size'=$VALUE).

Benefits

The adoption of the new KCQL syntax enhances the flexibility of the data-lakes sink connectors, empowering users to tailor configurations more precisely to their requirements. By transitioning keywords to entries in the PROPERTIES section, potential misconfigurations stemming from keyword order discrepancies are mitigated, ensuring configurations are applied as intended.

Upgrades

Please note that the upgrades to the data-lakes sink connectors are not backward compatible with existing configurations. Users are required to update their configurations to align with the new KCQL syntax and PROPERTIES entries. This upgrade is necessary for any instances of the sink connector (S3, Azure, GCP) set up before version 7.0.0.

To upgrade to the new version, users must follow these steps:

Stop the sink connectors.
Update the connector to version 7.0.0.
Edit the sink connectors’ KCQL setting to translate to the new syntax and PROPERTIES entries.

6.3.1

This update specifically affects datalake sinks employing the JSON storage format. It serves as a remedy for users who have resorted to a less-than-ideal workaround: employing a Single Message Transform (SMT) to return a Plain Old Java Object (POJO) to the sink. In such cases, instead of utilizing the Connect JsonConverter to seamlessly translate the payload to JSON, reliance is placed solely on Jackson.

However, it’s crucial to note that this adjustment is not indicative of a broader direction for future expansions. This is because relying on such SMT practices does not ensure an agnostic solution for storage formats (such as Avro, Parquet, or JSON).

6.3.0

NEW: HTTP Sink Connector

HTTP Sink Connector (Beta)

NEW: Azure Event Hubs Source Connector

Azure Event Hubs Source Connector.

All Connectors:

Update of dependencies, CVEs addressed.

Please note that the Elasticsearch 6 connector will be deprecated in the next major release.

6.2.0

NEW: GCP Storage Source Connector

GCP Storage Source Connector (Beta)

AWS Source Connector

Important: The AWS Source Partition Search properties have changed for consistency of configuration. The properties that have changed for 6.2.0 are:
- connect.s3.partition.search.interval changes to connect.s3.source.partition.search.interval.

JMS Source and Sink Connector

Jakarta Dependency Migration: Switch to Jakarta EE dependencies in line with industry standards to ensure evolution under the Eclipse Foundation.

All Connectors:

There has been some small tidy up of dependencies, restructuring and removal of unused code, and a number of connectors have a slimmer file size without losing any functionality.
The configuration directory has been removed as these examples are not kept up-to-date. Please see the connector documentation instead.
Dependency upgrades.

6.1.0

All Connectors:

In this release, all connectors have been updated to address an issue related to conflicting Antlr jars that may arise in specific environments.

AWS S3 Source:

Byte Array Support: Resolved an issue where storing the Key/Value as an array of bytes caused compatibility problems due to the connector returning java.nio.ByteBuffer while the Connect framework’s ByteArrayConverter only works with byte[]. This update ensures seamless conversion to byte[] if the key/value is a ByteBuffer.

JMS Sink:

Fix for NullPointerException: Addressed an issue where the JMS sink connector encountered a NullPointerException when processing a message with a null JMSReplyTo header value.

JMS Source:

Fix for DataException: Resolved an issue where the JMS source connector encountered a DataException when processing a message with a JMSReplyTo header set to a queue.

AWS S3 Sink/GCP Storage Sink (beta)/Azure Datalake Sink (beta):

GZIP Support for JSON Writing: Added support for GZIP compression when writing JSON data to AWS S3, GCP Storage, and Azure Datalake sinks.

6.0.2

GCP Storage Sink (beta):

Improve suppport for handling GCP naming conventions

6.0.1

AWS S3 / Azure Datalake (Beta) / GCP Storage (Beta)

Removed check preventing nested paths being used in the sink.
Avoid cast exception in GCP Storage connector when using Credentials mode.

6.0.0

New Connectors introduced in 6.0.0:

Azure Datalake Sink Connector (Beta)
GCP Storage Sink Connector (Beta)

Deprecated Connectors removed in 6.0.0:

Kudu
Hazelcast
Hbase
Hive

All Connectors:

Standardising package names. Connector class names and converters will need to be renamed in configuration.
Some clean up of unused dependencies.
Introducing cloud-common module to share code between cloud connectors.

5.0.1

New Connectors introduced in 5.0.1:

Consumer Group Offsets S3 Sink Connector

AWS S3 Connector - S3 Source & Sink

Enhancement: BigDecimal Support

Redis Sink Connector

Bug fix: Redis does not initialise the ErrorHandler

5.0.0

All Connectors

Test Fixes and E2E Test Clean-up: Improved testing with bug fixes and end-to-end test clean-up.
Code Optimization: Removed unused code and converted Java code and tests to Scala for enhanced stability and maintainability.
Ascii Art Loading Fix: Resolved issues related to ASCII art loading.

AWS S3 Connector - S3 Source & Sink

The source and sink has been the focus of this release.

Full message backup. The S3 sink and source now supports full message backup. This is enabled by adding in the KCQL PROPERTIES('store.envelope'=true)
Removed Bytes_*** storage format. For those users leveraging them there is a migration information below. Storing raw Kafka message the storage format should be AVRO/PARQUET/JSON(less ideal).
Introduced support for BYTES storing single message as raw binary. Typically, storing images or videos are the use case for this. This is enabled by adding in the KCQL STOREAS BYTES

Sink

Enhanced PARTITIONBY Support: expanded support for PARTITIONBY fields, now accommodating fields containing dots. For instance, you can use PARTITIONBY a, `field1.field2` for enhanced partitioning control.
Advanced Padding Strategy: a more advanced padding strategy configuration. By default, padding is now enforced, significantly improving compatibility with S3 Source.
Improved Error Messaging: Enhancements have been made to error messaging, providing clearer guidance, especially in scenarios with misaligned topic configurations (#978).

Source

Expanded Text Reader Support: new text readers to enhance data processing flexibility, including:
- Regex-Driven Text Reader: Allows parsing based on regular expressions.
- Multi-Line Text Reader: Handles multi-line data.

Kudu and Hive

The Kudu and Hive connectors are now deprecated and will be removed in a future release.

InfluxDB

Fixed a memory issue with the InfluxDB writer.
Upgraded to Influxdb2 client (note: doesn’t yet support Influxdb2 connections).

S3 upgrade notes

Upgrading from 5.0.0 (preview) to 5.0.0

For installations that have been using the preview version of the S3 connector and are upgrading to the release, there are a few important considerations:

Previously, default padding was enabled for both “offset” and “partition” values starting in June.

However, in version 5.0, the decision to apply default padding to the “offset” value only, leaving the " partition" value without padding. This change was made to enhance compatibility with querying in Athena.

If you have been using a build from the master branch since June, your connectors might have been configured with a different default padding setting.

To maintain consistency and ensure your existing connector configuration remains valid, you will need to use KCQL configuration properties to customize the padding fields accordingly.

Upgrading from 4.x to 5.0.0

Starting with version 5.0.0, the following configuration keys have been replaced.

Field

Old Property

New Property

**Upgrading from 4.1.* and 4.2.0**

In version 4.1, padding options were available but were not enabled by default. At that time, the default padding length, if not specified, was set to 8 characters.

However, starting from version 5.0, padding is now enabled by default, and the default padding length has been increased to 12 characters.

Enabling padding has a notable advantage: it ensures that the files written are fully compatible with the Lenses Stream Reactor S3 Source, enhancing interoperability and data integration.

Sinks created with 4.2.0 and 4.2.1 should retain the padding behaviour, and, therefore should disable padding:

If padding was enabled in 4.1, then the padding length should be specified in the KCQL statement:

Upgrading from 4.x to 5.0.0 only when `STOREAS Bytes_*` is used**

The Bytes_*** storage format has been removed. If you are using this storage format, you will need to install the 5.0.0-deprecated connector and upgrade the connector instances by changing the class name:

Source Before:

Source After:

Sink Before:

Sink After:

The deprecated connector won’t be developed any further and will be removed in a future release. If you want to talk to us about a migration plan, please get in touch with us at .

Upgrade a connector configuration

To migrate to the new configuration, please follow the following steps:

stop all running instances of the S3 connector
upgrade the connector to 5.0.0
update the configuration to use the new properties

4.2.0

All
- Ensure connector version is retained by connectors
- Lenses branding ASCII art updates

4.1.0

Note From Version 4.10, AWS S3 Connectors will use the AWS Client by default. You can revert to the jClouds version by setting the connect.s3.aws.client property.

All
- Scala upgrade to 2.13.10
- Dependency upgrades

4.0.0

All
- Scala 2.13 Upgrade
- Gradle to SBT Migration

3.0.1

All
- Replace Log4j with Logback to overcome CVE-2021-44228
- Bringing code from legacy dependencies inside of project

3.0.0

All
- Move to KCQL 2.8.9
- Change sys.errors to ConnectExceptions

2.1.3

Move to connect-common 2.0.5 that adds complex type support to KCQL

2.1.2

AWS S3 Sink Connector
- Prevent null pointer exception in converters when maps are presented will null values
- Offset reader optimisation to reduce S3 load

2.1.0

AWS S3 Sink Connector
Elasticsearch 7 Support

2.0.1

Hive Source
- Rename option connect.hive.hive.metastore to connect.hive.metastore
- Rename option connect.hive.hive.metastore.uris

2.0.0

Move to Scala 2.12
Move to Kafka 2.4.1 and Confluent 5.4
Deprecated:

Kafka Connectors

Overview

Understanding Kafka Connect

hashtagKafka Connect Core Concepts

hashtagConnectors Plugins

hashtagTasks vs Connectors

hashtagConverters

hashtagSecret Providers

hashtagSingle Message Transforms (SMTs)

Connectors

Install

Sources

Sinks

hashtagFAQ

hashtagCan the datalakes sinks lose data?

hashtagDo the datalake sinks support exactly once semantics?

hashtagHow do I escape dots in field names in KCQL?

hashtagHow do I escape other special characters in field names in KCQL?

Azure Event Hubs

Secret Providers

hashtag

Install

hashtagSecret Rotation

Environment Provider

hashtagConfiguration

hashtagUsage

hashtagData encoding

AES256

hashtagConfiguring the plugin

hashtagUsage

Single Message Transforms

Overview

hashtagInstallation

hashtagSMT installation on Lenses platform

hashtagSMT installation on MSK connect

InsertRecordTimestampHeaders

InsertRollingWallclock

hashtagTransform Type Class

hashtag

InsertSourcePartitionOrOffsetValue

InsertWallclock

InsertWallclockDateTimePart

hashtagTransform Type Class

hashtagConfiguration

hashtagExample

Tutorials

Creating & managing a connector

hashtagCreating your Kafka Connector Configuration

Cloud Storage Examples

hashtag

Http Sink Templating

hashtagStatic string template

hashtag

Sink converters & different data formats

hashtagSchema.Struct and a Struct Payload

hashtagSchema.String and a JSON Payload

hashtagNo schema and a JSON Payload

Source converters with incoming JSON or Avro

Loading XML from Cloud storage

Loading ragged width files

Using the MQTT Connector with RabbitMQ

Using Error Policies

Using dead letter queues

Contributing

hashtagHow to contribute?

Developing a connector

hashtagSetting up a new module

Utilities

Testing

Downloads

Release notes

Secret Providers

hashtag2.3.0

hashtagAES256 Provider:

hashtagAWS Secrets Manager Provider:

hashtagAzure Secret Provider:

hashtagHashicorp Vault Provider:

Overview

Azure Event Hubs

Cloud Storage Examples

Kafka Connect Core Concepts

Connectors Plugins

Tasks vs Connectors

Converters

Secret Providers

Single Message Transforms (SMTs)

FAQ

Can the datalakes sinks lose data?

Do the datalake sinks support exactly once semantics?

How do I escape dots in field names in KCQL?

How do I escape other special characters in field names in KCQL?

Secret Rotation

Configuration

Usage

Data encoding

Configuring the plugin

Usage

Installation

SMT installation on Lenses platform

SMT installation on MSK connect

Transform Type Class

Transform Type Class

Configuration

Example

Creating your Kafka Connector Configuration

Static string template

Schema.Struct and a Struct Payload

Schema.String and a JSON Payload

No schema and a JSON Payload

How to contribute?

Setting up a new module

2.3.0

AES256 Provider:

AWS Secrets Manager Provider:

Azure Secret Provider:

Hashicorp Vault Provider:

Secret Rotation

Kafka Connect Core Concepts

Connectors Plugins

Tasks vs Connectors

Converters

Secret Providers

Single Message Transforms (SMTs)

Installation

SMT installation on Lenses platform

SMT installation on MSK connect

How to contribute?

Static string template

Throw

Noop

Retry

AvroConverter

JsonPassThroughConverter

JsonSimpleConverter

JsonConverterWithSchemaEvolution

BytesConverter

Dynamic string template containing json message fields

Dynamic string template containing whole json message

Dynamic string template containing avro message fields

FAQ

Can the datalakes sinks lose data?

Do the datalake sinks support exactly once semantics?

How do I escape dots in field names in KCQL?

How do I escape other special characters in field names in KCQL?

Transform Type Class

Configuration

Example

Configuring the plugin

Usage

Configuration

Usage

Data encoding

2.3.0

AES256 Provider:

AWS Secrets Manager Provider:

Azure Secret Provider: