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.

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.

Azure Event Hubs

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

A Kafka Connect source connector to read events from Azure Event Hubs and push them to Kafka.

In order to leverage Kafka API in your Event Hubs it has to be at least on Standard Pricing Tier. More info here.

Connector Class

Example

For more examples see the .

Below example presents all the necessary parameters configuration in order to use Event Hubs connector. It contains all the necessary parameters (but nothing optional, so feel free to tweak it to your needs):

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 following KCQL is supported:

The selection of fields from the Event Hubs message is not supported.

Payload support

As for now Azure Event Hubs Connector supports raw bytes passthrough from source Hub to Kafka Topic specified in the KCQL config.

Authentication

You can connect to Azure EventHubs passing specific JAAS parameters in configuration.

Learn more about different methods of connecting to Event Hubs on . The only caveat is to add connector-specific prefix like in example above. See for more info.

Fine-tunning the Kafka Connector

The Azure Event Hubs Connector utilizes the Apache Kafka API implemented by Event Hubs. This also allows fine-tuning for user-specific needs because the Connector passes all of the properties with a specific prefix directly to the consumer. The prefix is connect.eventhubs.connection.settings and when user specifies a property with it, it will be automatically passed to the Consumer.

User wants to fine-tune how much data records comes through the network at once. He specifies below property as part of his configuration for Azure Event Hubs Connector before starting it.

It means that internal Kafka Consumer will poll at most 100 records at time (as max.poll.records is passed directly to it)

There are certain exceptions to this rule as couple of those are internally used in order to smoothly proceed with consumption. Those exceptions are listed below:

client.id - Connector sets it by itself
group.id - Connector sets it by itself
key.deserializer - Connector transitions bytes 1-to-1

Option Reference

Name

Description

Type

Default Value

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!

InfluxDB

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

This connector has been retired starting version 11.0.0

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.

Azure KeyVault

This page describes how to retrieve secrets from Azure KeyVault for use in Kafka Connect.

Secure secrets in Azure KeyVault and use them in Kafka Connect.

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:

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

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 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 . Furthermore, they support 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)

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

io.lenses.connect.smt.header.InsertRollingWallclock

Configuration

Name

Description

Type

Default

Valid Values

Importance

Example

To store the epoch value, use the following configuration:

To store a string representation of the date and time in the format yyyy-MM-dd HH:mm:ss.SSS, use the following:

To use the timezone Asia/Kolkoata, use the following:

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.

Configuration Properties

Configuration Property

Description

Optionality

Default Value

Default Value: Specifies the default value assigned if no value is explicitly provided in the configuration.

These properties allow you to customize which fields from the offset and partition of a SourceRecord are added as headers, along with specifying optional prefixes for the header keys. Adjust these configurations based on your specific use case and data requirements.

Example Configuration

transforms: This property lists the transformations to be applied to the records.
transforms.InsertSourcePartitionOrOffsetValue.type: Specifies the class implementing the transformation (InsertSourcePartitionOrOffsetValue in this case).
transforms.InsertSourcePartitionOrOffsetValue.offset.fields

Example Usage with Cloud Connectors

AWS S3, Azure Datalake or GCP Storage

When using this transformation with AWS S3, you can configure your Kafka Connect connector as follows:

To customise the header prefix you can also set the header values:

Replace path,line,ts and container,prefix with the actual field names you are interested in extracting from the partition or offset.

By using InsertSourcePartitionOrOffsetValue transformation, you can enrich your data records with additional metadata headers based on partition or offset values before they are delivered to your cloud storage destinations.

Using the Prefix Feature in InsertSourcePartitionOrOffsetValue Transformation

The prefix feature in InsertSourcePartitionOrOffsetValue allows you to prepend a consistent identifier to each header key added based on partition or offset values from SourceRecords.

Configure the transformation in your Kafka Connect connector properties:

offset.prefix: Specifies the prefix for headers derived from offset values. Default is "offset.".
partition.prefix: Specifies the prefix for headers derived from partition values. Default is "partition.".

By setting offset.prefix=offset. and partition.prefix=partition., headers added based on offset and partition fields will have keys prefixed accordingly in the SourceRecord headers.

This configuration ensures clarity and organization when inserting metadata headers into your Kafka records, distinguishing them based on their source (offset or partition). Adjust prefixes (offset.prefix and partition.prefix) as per your naming conventions or requirements.

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.

InsertWallclockHeaders

A Kafka Connect Single Message Transform (SMT) that inserts date, year, month,day, hour, minute and second headers using the system clock as a message header.

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

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

IDE Support (JSON Schemas)

Add support for creating connector configuration in your IDE.

The Connect JSON Schema toot is a tool that generates JSON Schema from Kafka Connect connector, Single Message Transforms and Secret Provider configurations. It helps validate and document connector configurations by providing a standardized schema format that can be used for validation and documentation purposes.

Many IDE's support JSON schemas allowing for intellisense and auto completion. For VS code you can find out more information here.

Generates JSON Schema from any Kafka Connect Connector, SMT or Secret Provider JAR
Includes all connector configuration fields with types and descriptions
Includes default Kafka Connect fields.

Beta - The Schemas are still beta and can change and are currently for the latest version of the Stream Reactor connector.

The Generator Tool will generate JSON schemas from jars containing classes implementing:

Sink Connector
Source Connector
Single Message Transforms
Secret Providers

In addition it can also merge the individual connectors schemas into a Union schema.

Optionally it can then include overrides for Connect converters (header, key & value) and consumer and producers.

Usage

To triggered the default snippets at the root level start typing:

Connectors: for connector examples
Overrides: for converters, consumer & producer

You can type further to get default snippets for different connector types, e.g. AWS S3.

VS Code Setup

Once you have JSON schema you can enable JSON Schema validation and auto completion in VS Code for both JSON and YAML files:

Install the required VS Code extensions

"YAML" by Red Hat
"JSON Language Features" (built-in)

This setup will provide:

Schema validation for both JSON and YAML files
Auto completion for connector configurations
Hover documentation for fields
Formatting support for YAML files

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 plugin.path
Each connector has mandatory configurations that must be deployed and validated; other configurations are optional. Always read the connector documentation first.
You can deploy the connector using Kafka Connect API or Lenses to manage it for you.

Sample of Connector, AWS S3 Sink Connector from Stream Rector:

Let's drill down this connector configuration and what will happen when I deploy:

connector.class is the plugin we will use.
task.max is how many tasks will be executed in the Kafka Connect Cluster. In this example, we will have 1 task; my topic has 9 partitions, so in this case, in the consumer group of this connector, we will have 1 instance. This one task will consume 9 partitions. To scale is just to increase the number of tasks.

value.converter is the format type used to deserialize or serialize the data in value. In this case, our value will be in json format.
key.converter is the format type used to deserialize or serialize the data in key. In this case, our key will be in string format.

connect.s3.aws.auth.mode is what is the type of authentication we will use to connect to the AWS S3 bucket.
connect.s3.aws.access.key is the access key to authenticate into the AWS S3 bucket.

For deep configuration of AWS S3 Sink connect, click

Managing your Connector

After deploying your Kafka Connector into your Kafka Connect Cluster, it will be managed by the Kafka Connect Cluster.

To better show how Kafka Connect manages your connectors, we will use Lenses UI.

The image below is the Lenses Connectors list:

In the following image, we will delve into the Kafka Connector details.

Consumer Group, when we use Lenses, we can see which consumer group is reading and consuming that topic.
Connector tasks, we can see which tasks are open, the status, and how many records are in and out of that topic.

Cloud Storage Examples

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

AWS S3 Source Examples

Examples for AWS S3 Source Kafka Connector.

Parquet Configuration, Using Avro Value Converter

This connector configuration is designed for ingesting data from , into Apache Kafka.

name=aws-s3SourceConnectorParquet # this can be anything
connector.class=io.lenses.streamreactor.connect.aws.s3.source.S3SourceConnector
tasks.max=1
connect.s3.kcql=insert into $TOPIC_NAME select * from $BUCKET_NAME:$PREFIX_NAME STOREAS `parquet`
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8089
    connect.s3.aws.region=eu-west-2
    connect.s3.aws.secret.key=SECRET_KEY
    connect.s3.aws.access.key=ACCESS_KEY
    connect.s3.aws.auth.mode=Credentials

Connector Name: aws-s3SourceConnectorParquet (This can be customized as needed)
Connector Class: io.lenses.streamreactor.connect.aws.s3.source.S3SourceConnector
Maximum Tasks: 1 (Number of tasks to execute in parallel)
KCQL Statement:
Inserts data from the specified cloud storage bucket into a Kafka topic.
Syntax: insert into $TOPIC_NAME select * from $BUCKET_NAME:$PREFIX_NAME STOREAS 'parquet'
$TOPIC_NAME: Name of the Kafka topic where data will be inserted.
$BUCKET_NAME: Name of the AWS S3 storage bucket.
$PREFIX_NAME: Prefix or directory within the bucket.
Value Converter:
AvroConverter (Assuming data is serialized in Avro format)
Schema Registry URL:
http://localhost:8089 (URL of the schema registry for Avro serialization)
Authentication Properties:
(These properties depend on the authentication mechanism used for accessing the cloud storage service. Replace placeholders with actual authentication properties for the specific cloud platform.)

This configuration serves as a template and can be customized according to the requirements and specifics of your data.

Envelope Storage

This configuration example is particularly useful when you need to restore data from a AWS S3, into Apache Kafka while maintaining all data including headers, key and value for each record. The envelope structure encapsulates the actual data payload along with metadata into files on your source bucket, providing a way to manage and process data with additional context.

When to Use This Configuration:

Data Restoration with Envelope Structure: If you’re restoring data from AWS S3 into Kafka and want to preserve metadata, this configuration is suitable. Envelopes can include metadata like timestamps, data provenance, or other contextual information, which can be valuable for downstream processing.
Batch Processing: The configuration supports batch processing by specifying a batch size (BATCH=2000) and a limit on the number of records (LIMIT 10000). This is beneficial when dealing with large datasets, allowing for efficient processing in chunks.

This configuration ensures efficient and reliable data ingestion from cloud storage into Kafka while preserving the envelope structure and providing robust error handling mechanisms.

AVRO Configuration Envelope Storage

Similar to the above, this is another configuration for envelope format

When to Use This Configuration Variant:

Single Task Processing: Unlike the previous configuration which allowed for multiple tasks (tasks.max=4), this variant is configured for single-task processing (tasks.max=1). This setup may be preferable in scenarios where the workload is relatively lighter or where processing tasks in parallel is not necessary.
AVRO Format for Data Serialization: Data is serialized in the AVRO format (STOREAS 'AVRO'), leveraging the AvroConverter (value.converter=io.confluent.connect.avro.AvroConverter). This is suitable for environments where Avro is the preferred serialization format or where compatibility with Avro-based systems is required.

AWS S3 Sink Time Based Partitioning

Examples for AWS S3 Sink Kafka Connector time based partitioning.

Partitioning by Date and Time

This scenario partitions data by date and time, employing record timestamp headers to enable partitioning based on these time components.

GCP Source

Examples for GCP Source Kafka Connector.

Parquet Configuration, Using Avro Value Converter

This connector configuration is designed for ingesting data from , into Apache Kafka.

name=gcp-storageSourceConnectorParquet # this can be anything
connector.class=io.lenses.streamreactor.connect.gcp.storage.source.GCPStorageSourceConnector
tasks.max=1
connect.gcpstorage.kcql=insert into $TOPIC_NAME select * from $BUCKET_NAME:$PREFIX_NAME STOREAS `parquet`
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8089
    connect.gcpstorage.gcp.auth.mode=Credentials
    connect.gcpstorage.gcp.credentials=$GCP_CREDENTIALS
    connect.gcpstorage.gcp.project.id=$GCP_PROJECT_ID

Connector Name: gcp-storageSourceConnectorParquet (This can be customized as needed)
Connector Class: io.lenses.streamreactor.connect.gcp.storage.source.GCPStorageSourceConnector
Maximum Tasks: 1 (Number of tasks to execute in parallel)
KCQL Statement:
Inserts data from the specified cloud storage bucket into a Kafka topic.
Syntax: insert into $TOPIC_NAME select * from $BUCKET_NAME:$PREFIX_NAME STOREAS 'parquet'
$TOPIC_NAME: Name of the Kafka topic where data will be inserted.
$BUCKET_NAME: Name of the GCP Storage storage bucket.
$PREFIX_NAME: Prefix or directory within the bucket.
Value Converter:
AvroConverter (Assuming data is serialized in Avro format)
Schema Registry URL:
http://localhost:8089 (URL of the schema registry for Avro serialization)
Authentication Properties:
(These properties depend on the authentication mechanism used for accessing the cloud storage service. Replace placeholders with actual authentication properties for the specific cloud platform.)

This configuration serves as a template and can be customized according to the requirements and specifics of your data.

Envelope Storage

This configuration example is particularly useful when you need to restore data from a GCP Storage, into Apache Kafka while maintaining all data including headers, key and value for each record. The envelope structure encapsulates the actual data payload along with metadata into files on your source bucket, providing a way to manage and process data with additional context.

When to Use This Configuration:

Data Restoration with Envelope Structure: If you’re restoring data from GCP Storage into Kafka and want to preserve metadata, this configuration is suitable. Envelopes can include metadata like timestamps, data provenance, or other contextual information, which can be valuable for downstream processing.
Batch Processing: The configuration supports batch processing by specifying a batch size (BATCH=2000) and a limit on the number of records (LIMIT 10000). This is beneficial when dealing with large datasets, allowing for efficient processing in chunks.

This configuration ensures efficient and reliable data ingestion from cloud storage into Kafka while preserving the envelope structure and providing robust error handling mechanisms.

AVRO Configuration Envelope Storage

Similar to the above, this is another configuration for envelope format

When to Use This Configuration Variant:

Single Task Processing: Unlike the previous configuration which allowed for multiple tasks (tasks.max=4), this variant is configured for single-task processing (tasks.max=1). This setup may be preferable in scenarios where the workload is relatively lighter or where processing tasks in parallel is not necessary.
AVRO Format for Data Serialization: Data is serialized in the AVRO format (STOREAS 'AVRO'), leveraging the AvroConverter (value.converter=io.confluent.connect.avro.AvroConverter). This is suitable for environments where Avro is the preferred serialization format or where compatibility with Avro-based systems is required.

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

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.

Not all Connectors support the source converters. Check the option reference for your connector.

Before records are passed back to connect, they go through the converter if specified.

AvroConverter

The payload is an Avro message. In this case, you need to provide a path for the Avro schema file to be able to decode it.

JsonPassThroughConverter

The incoming payload is JSON, the resulting Kafka message value will be of type string and the contents will be the incoming JSON.

JsonSimpleConverter

The payload is a JSON message. This converter will parse the JSON and create an Avro record for it which will be sent to Kafka.

JsonConverterWithSchemaEvolution

An experimental converter for translating JSON messages to Avro. The Avro schema is fully compatible as new fields are added as the JSON payload evolves.

BytesConverter

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.

Utilities

Coming soon!

Testing

Coming soon!

Lenses Connectors Support

This page describes the enterprise support available for the Stream Reactor Kafka Connect connectors.

Lenses offers support for our connectors as a paid subscription, The list of connectors is always growing and . Please reach out to for for any inquiries.

varies depending on connectors. The most recent connectors are considered "Next-Gen" and are available with a premium plus SLA to accommodate the most demanding requirements.

Support Level Available

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

Single Message Transforms

This page contains the release notes for Single Message Transforms.

1.3.2

Adds support for multiple "from" patterns.

This converts the format.from.pattern field in the following SMTs:

InsertFieldTimestampHeaders
InsertRollingFieldTimestampHeaders
TimestampConverter

into a List (comma separated) so that these SMTs can support multiple (fallback) DateTimeFormatter patterns should multiple timestamps be in use.

Configuration Compatibility

When updating your configuration, if format.from.pattern contains commas, enclose the pattern in double quotes.

Configurations should be backwards-compatible with previous versions of the SMT, the exception is if commas are used in the format.from.pattern string.

To update the configuration of format.from.pattern ensure you enclose any pattern which contains commas in double quotes.

Old Configuration:

New Configuration

Multiple format.from.pattern can now be configured, each pattern containing a comma can be enclosed in double quotes:

Configuration Order

When configuring format.from.pattern, the order matters; less granular formats should follow more specific ones to avoid data loss. For example, place yyyy-MM-dd after yyyy-MM-dd'T'HH:mm:ss to ensure detailed timestamp information isn't truncated.

1.3.1

Increase error information for debugging.

1.3.0

Adds support for adding metadata to kafka connect headers (for a source connector).

1.2.1

Workaround for Connect runtime failing with unexplained exception where it looks like the static fields of parent class is not resolved prop.

1.2.0

Adds support for inserting time based headers using a Kafka message payload field.

1.1.2

Fix public visibility of rolling timestamp headers.

1.1.1

Don't make CTOR protected.

1.1.0

Introducing four new Single Message Transforms (SMTs) aimed at simplifying and streamlining the management of system or record timestamps, along with support for rolling windows. These SMTs are designed to significantly reduce the complexity associated with partitioning data in S3/Azure/GCS Sink based on time, offering a more efficient and intuitive approach to data organization. By leveraging these SMTs, users can seamlessly handle timestamp-based partitioning tasks, including optional rolling window functionality, paving the way for smoother data management workflows.

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.

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

io.lenses.streamreactor.connect.datalake.sink.DatalakeSinkConnector

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

GCP Storage

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

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.

Connector Class

Example

For more examples see the .

KCQL Support

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

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

Source Topic

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 configurability 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 object names, a practice that proves beneficial when using the GCP Storage Source connector to restore data. This object name padding ensures that objects are ordered lexicographically, allowing the GCP Storage Source connector to skip the need for reading, sorting, and processing all objects, thereby enhancing efficiency.

Partitioning & Object Keys

The object key serves as the filename used to store data in GCP Storage. 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 (GCP Athena 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 partition clause works on header, key and values fields of the Kafka message.

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 GCP Storage object key name:

Kafka message headers can also be used in the GCP Storage 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 GCP Storage object keys effectively.

To enable Athena-like partitioning, use the following syntax:

Rolling Windows

Storing data in GCP Storage and partitioning it by time is a common practice in data management. For instance, you may want to organize your GCP Storage 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 .

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 GCP Storage. 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 object. This feature proves particularly valuable for scenarios involving the storage of images or other binary data in GCP Storage. For cases where you prefer to consolidate multiple records into a single binary object, AVRO or Parquet are the recommended choices.

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 GCP Storage is desired.

Examples

Storing the message Value Avro data as Parquet in GCP Storage:

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

Enabling the full message stored as JSON in GCP Storage:

Enabling the full message stored as AVRO in GCP Storage:

If the restore (see the GCP Storage 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 an object flush after a specified number of records have been written to it.
Flush by Size - initiates an object flush once a predetermined size (in bytes) has been attained.
Flush by Interval - enforces an object 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 objects 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 object, with no new Kafka messages arriving for an extended period of 6 hours. To prevent undue delays, the interval flush guarantees that the object 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 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)

When connect.gcpstorage.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 object written to GCP Storage). 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 object.

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 GCP Storage 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 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. This is configured, as its name suggests, by default.

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. You can find detailed information on how to use the Connect Secret Providers . This approach ensures robust security practices while handling access credentials.

Error policies

The connector supports .

Indexes Prefix

The connector uses the concept of index objects 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 objects.

By default, the prefix for these index objects is named .indexes for all connectors. However, each connector will create and store its index objects within its own nested prefix inside this .indexes prefix.

You can configure the root prefix for these index objects using the property connect.gcpstorage.indexes.name. This property specifies the path from the root of the GCS bucket. Note that even if you configure this property, the connector will still create a nested prefix within the specified prefix.

Examples

Option Reference

Name

Description

Type

Available Values

Default Value

AWS S3

This page describes the usage of the Stream Reactor AWS S3 Sink Connector.

This Kafka Connect sink connector facilitates the seamless transfer of records from Kafka to AWS S3 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

io.lenses.streamreactor.connect.aws.s3.sink.S3SinkConnector

Example

For more examples see the .

This example writes to a bucket called demo, partitioning by a field called ts, store as JSON.

KCQL Support

The connector uses KCQL to map topics to S3 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

Here are a few examples:

SQL Projection

Source Topic

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

Partitioning & Object Keys

The object key serves as the filename used to store data in S3. 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 $bucket/[$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 $bucket/[$prefix]/$topic/customKey1=customValue1/customKey2=customValue2/topic(partition_offset).extension (AWS Athena naming style mimicking Hive-like data partitioning) or $bucket/[$prefix]/customValue/topic(partition_offset).ext. The extension is determined by the selected storage format.

The partition clause works on header, key and values fields of the Kafka message.

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 S3 object key name:

Kafka message headers can also be used in the S3 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 S3 object keys effectively.

To enable Athena-like partitioning, use the following syntax:

Rolling Windows

Storing data in Amazon S3 and partitioning it by time is a common practice in data management. For instance, you may want to organize your S3 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.

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 AWS S3. 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 S3. For cases where you prefer to consolidate multiple records into a single binary file, AVRO or Parquet are the recommended choices.

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 S3 is desired.

Examples

Storing the message Value Avro data as Parquet in S3:

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

Enabling the full message stored as JSON in S3:

Enabling the full message stored as AVRO in S3:

If the restore (see the S3 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).

The flush options are configured using the flush.count, flush.size, and flush.interval properties. The settings are optional and if not specified the defaults are:

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

When connect.s3.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 object written to S3). 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 object.

Properties

Name

Description

Type

Available Values

Default Value

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

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 S3 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 AWS authentication chain, simplifying the authentication process.
Credentials: In this mode, explicit configuration of AWS Access Key and Secret Key is required for authentication.

Here’s an example configuration for the Credentials mode:

For enhanced security and flexibility when using the Credentials mode, it is highly advisable to utilize Connect Secret Providers.

Error policies

The connector supports .

API Compatible systems

The connector can also be used against API compatible systems provided they implement the following:

Indexes Directory

By default, the index objects are grouped within a prefix named .indexes for all connectors. However, each connector will create and store its index objects within its own nested prefix inside this .indexes prefix.

You can configure the prefix for these index objects using the property connect.s3.indexes.name. This property specifies the path from the root of the S3 bucket. Note that even if you configure this property, the connector will still place the indexes within a nested prefix of the specified prefix.

Examples

Option Reference

Name

Description

Type

Available Values

Default Value

Stream Reactor

This page contains the release notes for the Stream Reactor.

11.6.2

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

Resolved fatal exception during worker pod restarts: Fixed an issue where the connector would crash with a fatal ConnectException: Index not found following a consumer group rebalance when worker pods were restarted. Previously, internal state synchronization issues during a restart could lead to corrupted data tracking and cause subsequent writes to Google Cloud Storage (GCS) to fail (resulting in 412 Precondition Failed errors). The state management and recovery logic have been hardened to ensure seamless, reliable operation during pod restarts and rebalances.

10.0.4

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

See notes from .

11.6.1

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

In this release, we've fixed a bug where failed uploads would trigger retries without clearing staged files, preventing disk space issues and ensuring continuous connector operation.

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.
Introduced Bulk Processing Mode to enhance write throughput.

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 The minimum duration in milliseconds for the first backoff

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 Map[TopicPartition, Offset]

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
Fix: Java class java.util.Date support in cloud sink maps

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.
Restart the sink connectors.

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.
- connect.s3.partition.search.continuous

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
resume the stopped connectors

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
- Upgrade to Kafka 3.3.0

4.0.0

All
- Scala 2.13 Upgrade
- Gradle to SBT Migration
- Producing multiple artifacts supporting both Kafka 2.8 and Kafka 3.1.

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
- Additional testing with TestContainers

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 to

2.0.0

Move to Scala 2.12
Move to Kafka 2.4.1 and Confluent 5.4
Deprecated:
- Druid Sink (not scala 2.12 compatible)

Kafka Connectors

Overview

Understanding Kafka Connect

hashtagKafka Connect Core Concepts

Connectors

Install

Sources

Azure Event Hubs

hashtagConnector Class

hashtagExample

hashtagKCQL support

hashtagPayload support

hashtagAuthentication

hashtagFine-tunning the Kafka Connector

hashtagOption Reference

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

InfluxDB

hashtagThis connector has been retired starting version 11.0.0

Secret Providers

Install

hashtagSecret Rotation

Azure KeyVault

Environment Provider

hashtagConfiguration

hashtagUsage

hashtagData encoding

AES256

hashtagConfiguring the plugin

hashtagUsage

Single Message Transforms

Overview

InsertRecordTimestampHeaders

InsertRollingWallclock

hashtagTransform Type Class

hashtagConfiguration

hashtagExample

InsertSourcePartitionOrOffsetValue

hashtagConfiguration Properties

hashtagExample Configuration

hashtagExample Usage with Cloud Connectors

hashtagAWS S3, Azure Datalake or GCP Storage

hashtagUsing the Prefix Feature in InsertSourcePartitionOrOffsetValue Transformation

InsertWallclock

InsertWallclockHeaders

InsertWallclockDateTimePart

IDE Support (JSON Schemas)

hashtagUsage

hashtagVS Code Setup

Tutorials

Creating & managing a connector

hashtagCreating your Kafka Connector Configuration

hashtagManaging your Connector

Cloud Storage Examples

AWS S3 Source Examples

hashtagParquet Configuration, Using Avro Value Converter

hashtagEnvelope Storage

hashtagWhen to Use This Configuration:

hashtagAVRO Configuration Envelope Storage

hashtagWhen to Use This Configuration Variant:

AWS S3 Sink Time Based Partitioning

hashtagPartitioning by Date and Time

GCP Source

hashtagParquet Configuration, Using Avro Value Converter

hashtagEnvelope Storage

hashtagWhen to Use This Configuration:

hashtagAVRO Configuration Envelope Storage

hashtagWhen to Use This Configuration Variant:

Http Sink Templating

hashtagStatic string template

Sink converters & different data formats

hashtagSchema.Struct and a Struct Payload

Source converters with incoming JSON or Avro

hashtagAvroConverter

hashtagJsonPassThroughConverter

Kafka Connect Core Concepts

Connector Class

Example

KCQL support

Payload support

Authentication

Fine-tunning the Kafka Connector

Option Reference

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?

This connector has been retired starting version 11.0.0

Secret Rotation

Configuration

Usage

Data encoding

Configuring the plugin

Usage

Transform Type Class

Configuration

Example

Configuration Properties

Example Configuration

Example Usage with Cloud Connectors

AWS S3, Azure Datalake or GCP Storage

Using the Prefix Feature in InsertSourcePartitionOrOffsetValue Transformation

Usage

VS Code Setup

Creating your Kafka Connector Configuration

Managing your Connector

Parquet Configuration, Using Avro Value Converter

Envelope Storage

When to Use This Configuration:

AVRO Configuration Envelope Storage

When to Use This Configuration Variant:

Partitioning by Date and Time

Parquet Configuration, Using Avro Value Converter

Envelope Storage

When to Use This Configuration:

AVRO Configuration Envelope Storage

When to Use This Configuration Variant:

Static string template

Schema.Struct and a Struct Payload

AvroConverter

JsonPassThroughConverter

JsonSimpleConverter

JsonConverterWithSchemaEvolution

BytesConverter

How to contribute?

Support Level Available

2.3.0

1.3.2

Configuration Compatibility

Old Configuration:

New Configuration

Configuration Order

1.3.1

1.3.0

1.2.1

1.2.0

1.1.2

1.1.1

1.1.0

Secret Rotation

Kafka Connect Core Concepts

Static string template

Schema.Struct and a Struct Payload

Dynamic string template containing json message fields

Dynamic string template containing whole json message

Dynamic string template containing avro message fields