Overview

The OLake Go Kafka Source connector syncs messages from Kafka topics directly to the destination. It supports only one synchronization mode and offers features like parallel partition processing and checkpointing. This connector can be used within the OLake UI or run locally via Docker for open-source workflows.

Important: Kafka Retention Period

Kafka topics have a configured retention period. Once messages exceed this retention period, they are automatically deleted from the topic. OLake Go cannot access or sync deleted messages. Ensure your sync schedule runs frequently enough to capture data before it expires.

note

Kafka Connector Requirements

Kafka as a source connector is available in OLake Go v0.3.0 and above on both the source and destination side.
The OLake Go Kafka connector has been tested specifically with Amazon MSK (Amazon Managed Streaming for Apache Kafka), AWS's fully managed Apache Kafka service. The connector uses standard Kafka protocols and should work with other Kafka distributions.

Prerequisites

Version Prerequisites

Kafka Version 2.4.0 or higher.

Connection Prerequisites

OLake Go and the Kafka broker servers must be accessible within the same network.

After initial Prerequisites are fulfilled, the configurations for Kafka can be configured.

Authentication

OLake Go supports three authentication methods for connecting to Kafka brokers:

1. PLAINTEXT

No authentication or encryption. Suitable only for development or testing environments where security is not a concern.

Security Protocol: PLAINTEXT
Configuration: No additional authentication parameters required

2. SASL_PLAINTEXT

Authentication without encryption. Credentials are verified, but data is transmitted in plaintext.

Security Protocol: SASL_PLAINTEXT
Supported SASL Mechanisms:
- PLAIN: Simple username/password authentication
- SCRAM-SHA-512: Salted Challenge Response Authentication Mechanism with SHA-512 hashing

Example Configuration:

json
"protocol": {
  "security_protocol": "SASL_PLAINTEXT",
  "sasl_mechanism": "SCRAM-SHA-512",
  "sasl_jaas_config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"YOUR_USERNAME\" password=\"YOUR_PASSWORD\";"
}

3. SASL_SSL

Authentication with SSL/TLS encryption. This is the most secure option, providing both authentication and encrypted communication.

Security Protocol: SASL_SSL
Supported SASL Mechanisms:
- PLAIN: Simple username/password authentication over SSL
- SCRAM-SHA-512: Salted Challenge Response Authentication Mechanism with SHA-512 hashing over SSL

Example Configuration:

json
"protocol": {
  "security_protocol": "SASL_SSL",
  "sasl_mechanism": "SCRAM-SHA-512",
  "sasl_jaas_config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"YOUR_USERNAME\" password=\"YOUR_PASSWORD\";"
}

Configuration

Use Olake UI for Kafka
Use OLake CLI for Kafka

1. Navigate to the Source Configuration Page

Complete the OLake UI Setup Guide
After logging in to the OLake UI, select the Sources tab from the left sidebar.
Click Create Source on the top right corner.
Select Kafka from the connector dropdown
Provide a name for this source.

2. Provide Configuration Details

Enter Kafka credentials.

Connection details

Field	Description	Example Value
Kafka BootstrapServers `required`	Comma-separated list of Kafka broker addresses (host:port) for establishing the initial connection to the Kafka cluster.	`"broker:9092, broker:9093"`
Consumer Group ID	Unique identifier for the consumer group used to track messages and coordinate with consumer members. If not provided, OLake Go automatically generates one.	`example-consumer-group`
Topic Pattern	Regular expression to filter which Kafka topics are discovered during source setup. Use this to include only topics matching a naming pattern, or exclude specific topics from discovery. If left blank, all topics are discovered.	`^orders-.` - Only topics starting with `orders-` are discovered `^(?!orders-).+` - Topics that do not start with `orders-` are discovered `^(?=.order)(?=.*product).+$` - Discovers topics with `order` and `product` in the name
Security Protocol `required`	Protocol used for communication with Kafka. Supported options: `PLAINTEXT`, `SASL_PLAINTEXT`, or `SASL_SSL`.	`SASL_SSL`
SASL Mechanism	Specifies the type of SASL authentication protocol used to verify client identity when connecting to Kafka. Supported options: `PLAIN`, `SCRAM-SHA-512`.	`"SCRAM-SHA-512"`
SASL JAAS Config	JAAS configuration string containing the login module and authentication credentials (username and password) for SASL authentication.	If SASL mechanism is `PLAIN`: `"org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";"` If SASL mechanism is `SCRAM-SHA-512`: `"org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";"`
Schema Registry Endpoint	Endpoint of the Confluence Schema Registry. For AVRO-based topics, schema registry is mandatory.	`http://localhost:8081`
Authentication Type	Authentication method for the Schema Registry. Supported options: - `No Authentication` - `Username & Password` - `Bearer Token`	1. `No Authentication` - nothing is required 2. `Username & Password` - username and password is required (e.g., `username` = dummy, `password` = password) 3. `Bearer Token` - bearer token is required (e.g., dummy_token)

Additional configuration

Field	Description	Example Value
Threads Equal Total Partitions	When `true`, the number of consumer thread is equal to total number of partitions for most optimal parallel processing. When `false`, consumer threads are set to the `max_threads` value.	`false`
Max Threads	Maximum number of parallel threads for processing or syncing data.	`3`
Retry Count	Number of retry attempts for establishing sync with exponential backoff.	`3`

3. Test Connection

Once the connection is validated, the Kafka source is created. Jobs can then be configured using this source.
In case of connection failure, refer to the Troubleshooting section.

1. Create Configuration File

Once the OLake CLI is set up, create a folder to store configuration files such as source.json and destination.json.

The source.json file for Kafka must contain these mandatory fields.

2. Provide Configuration Details

An example source.json file will look like this:

source.json
{
    "bootstrap_servers": "broker:9092, broker:9093",
    "protocol": {
      "security_protocol": "SASL_SSL",
      "sasl_mechanism": "SCRAM-SHA-512",
      "sasl_jaas_config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"YOUR_KAFKA_USERNAME\" password=\"YOUR_KAFKA_PASSWORD\";"
    },
    "consumer_group_id": "example-consumer-group",
    "max_threads": 3,
    "threads_equal_total_partitions": false,
    "backoff_retry_count": 3,
    "topic_pattern": "^orders-.*",
    "schema_registry": {
      "endpoint": "http://localhost:8081",
      "username": "dummy",
      "password": "password"
  }
}

Field	Description	Example Value	Type
bootstrap_servers `required`	Comma-separated list of Kafka broker addresses (host:port) for establishing the initial connection to the Kafka cluster.	`"broker:9092, broker:9093"`	STRING
consumer_group_id	Unique identifier for the consumer group used to track message and coordinate with other consumer members. If not provided, OLake Go automatically generates one.	`example-consumer-group`	STRING
protocol `required`	Configuration object containing Kafka security settings and authentication configurations for connecting to Kafka. Sub-parameters: `security_protocol`, `sasl_mechanism`, `sasl_jaas_config`.	`{` `"security_protocol": "SASL_SSL",` `"sasl_mechanism": "SCRAM-SHA-512",` `"sasl_jaas_config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"YOUR_KAFKA_USERNAME\" password=\"YOUR_KAFKA_PASSWORD\";"` `}`	OBJECT
security_protocol `required`	Protocol used for communication with Kafka. Supported options: `PLAINTEXT`, `SASL_PLAINTEXT`, or `SASL_SSL`.	`"SASL_SSL"`	STRING
sasl_mechanism	Specifies the type of SASL authentication protocol used to verify client identity when connecting to Kafka. . Supported options: `PLAIN`, `SCRAM-SHA-512`.	`"SCRAM-SHA-512"`	STRING
sasl_jaas_config	JAAS configuration string containing the login module and authentication credentials (username and password) for SASL authentication.	If SASL mechanism is `PLAIN`: `"org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";"` If SASL mechanism is `SCRAM-SHA-512`: `"org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";"`	STRING
threads_equal_total_partitions	When `true`, the number of consumer thread is equal to total number of partitions for most optimal parallel processing. When `false`, consumer threads are set to the `max_threads` value.	`false`	BOOLEAN
max_threads	Maximum number of parallel threads for processing or syncing data.	`3`	INTEGER
backoff_retry_count	Number of retry attempts for establishing sync with exponential backoff.	`3`	INTEGER
schema_registry `Optional`	If schema registry is configured, at least schema registry `endpoint` has to be provided. If authentication is also present, then either authenticate using `username` and `password` or `bearer_token`. For AVRO-based topics, schema registry is mandatory.	`"schema_registry": {` `"endpoint": "http://localhost:8081",` `"username": "dummy",` `"password": "dummy-password"` `}`	OBJECT
topic_pattern	Regular expression to filter which Kafka topics are discovered during source setup. Use this to include only topics matching a naming pattern, or exclude specific topics from discovery. If left blank, all topics are discovered.	`^orders-.` - Only topics starting with `orders-` are discovered `^(?!orders-).+` - Topics that do not start with `orders-` are discovered `^(?=.order)(?=.*product).+$` - Discovers topics with `order` and `product` in the name	STRING

Similarly, destination.json file can be created inside this folder. For more information, see destination documentation.

3. Check Source Connection

To verify the database connection following command needs to be run:

check command
docker run --pull=always \
-v "[PATH_OF_CONFIG_FOLDER]:/mnt/config" \
olakego/source-kafka:latest \
check \
--config /mnt/config/source.json

If OLake Go is able to connect with Kafka {"connectionStatus":{"status":"SUCCEEDED"},"type":"CONNECTION_STATUS"} response is returned.
In case of connection failure, refer to the Troubleshooting section.

Kafka General Information

When OLake Go syncs data from Kafka topics to Iceberg, it transfers the complete message structure along with Kafka metadata. This ensures full traceability and context for each message.

Consumer Group Requirements

The consumer group configured for OLake Go must be dedicated exclusively to OLake Go. Also each job must use a unique consumer group ID. This consumer group should not be shared with any other process.
If a consumer group rebalance occurs while a sync is in progress, the sync will shut down gracefully. This means that all messages read up to that point are written to the destination, and the next sync resumes from where it left off.

Columns Transferred

The following columns are automatically included when syncing Kafka messages to the destination:

Column	Description
_kafka_key	The key of the message.
_kafka_offset	The sequential position of the message within its partition. Guarantees message ordering within a partition.
_kafka_partition	The Kafka partition number where the message resides. Used for parallel processing and data distribution.
_kafka_timestamp	The timestamp when the message was produced or appended to the Kafka topic.

Message Format Requirement

The Kafka messages must contain data in JSON or AVRO format.
For AVRO format, a Confluence Schema Registry must be configured—schema registry is mandatory only for AVRO data and if not provided OLake Go will consider the messages as JSON format.
Null or empty messages are handled based on the stream's ingestion mode:
- Append: the message is read but not written to the destination.
- Upsert: behavior depends on the configured Deduplication Key and whether the message value is null. See Upsert Behavior for detailed scenarios.

Normalization

OLake Go provides flexibility in how message values are stored:

Normalization = true: The payload in the value column is Level 0 flattened, and each nested field becomes a separate column in the destination table. This makes querying individual fields easier.
Normalization = false: The payload is stored as-is in the value column without flattening. The entire JSON object remains in a single column.

Stream Configuration

When Upsert is selected as ingestion mode for a stream(topic), a Deduplication Key must be configured in the Stream Configuration section. This key uniquely identifies which records represent the same entity in your Kafka topic, ensuring only the latest state is maintained in the destination.

info

For deduplication key use a field that will not change or get modified to prevent duplicates.

For example, if your topic receives multiple updates for customer_id = 101, the deduplication key ensures only the most recent version is stored, preventing duplicates of the same customer in your Iceberg table.

Multiple columns can be selected as deduplication keys
You can include _kafka_key and any columns from the message payload
Kafka Upsert ingestion mode only supports Positional Deletes as upsert type.

Partition Count Changes

Do not increase the number of partitions for a Kafka topic while an OLake Go sync job is running in Upsert mode. When partition count changes, messages with the same key will route to different partitions. For example, a key that hashed to partition 2 with 6 partitions might hash to partition 4 with 7 partitions. Since existing messages remain in their original partitions while new messages route elsewhere, concurrent processing can lead to duplicate records or incorrect final state.

Upsert Behavior

In Upsert mode, OLake Go uses the configured Deduplication Key to decide whether an incoming message is a new record, an update, or a delete. The operation depends on where the Deduplication Key comes from.

Deduplication Key is `_kafka_key`

Scenario	Behavior
Dedupe key is null or missing and the message value is present	The record gets appended
Dedupe key is empty bytes and the message value is present	The record gets upserted
Dedupe key is present and the message value is present	The record gets upserted
Dedupe key is present and the message value is null	The matching record is deleted
Dedupe key is null/missing and the message value is null (tombstone)	The record is skipped

Multiple Deduplication Keys (including `_kafka_key` plus value fields)

Scenario	Behavior
All configured dedupe key fields are missing/ JSON null/ mix of missing and null dedupe key fields	The sync fails
Single dedupe key, which is from the message value	The record gets upserted
One dedupe key field is an empty string	The record gets upserted
Multiple dedupe key fields configured, at least one is non-null	The record gets upserted
Conflict between dedupe key in message value and JSON value in key field (e.g. both having `id` as dedupe key)	The record gets upserted (body value takes precedence)

note

If multiple deduplication keys are selected and at least one of the keys is non-null, for the sync will run successfully.
Tombstone deletes are only supported for single deduplication key which is _kafka_key.

Version Requirements

For using Upsert mode in Kafka, the following version requirements must be met:

OLake Go Version 0.12.1 and above
OLake UI Version 0.7.1 and above

Troubleshooting

1. Consumer Group Corruption

If a consumer group becomes corrupted, you'll need to use a different consumer group to continue syncing data. The resolution depends on how you're running OLake Go.

Using OLake UI

Consumer group names cannot be modified once set in the OLake UI. To resolve a corrupted consumer group, create a new job with a different consumer group ID.

Using OLake CLI

When using the CLI, you can modify the consumer group in your configuration file:

Open your source.json file
Update the consumer_group_id field with a new consumer group name.
Save the file and run the sync command again

The sync will continue with the new consumer group.

If the issue is not listed here, post the query on Slack to get it resolved within a few hours.

Prerequisites​

Version Prerequisites​

Connection Prerequisites​

Authentication​

1. PLAINTEXT​

2. SASL_PLAINTEXT​

3. SASL_SSL​

Configuration​

1. Navigate to the Source Configuration Page​

2. Provide Configuration Details​

Connection details​

Additional configuration​

3. Test Connection​

1. Create Configuration File​

2. Provide Configuration Details​

3. Check Source Connection​

Kafka General Information​

Columns Transferred​

Normalization​

Stream Configuration​

Upsert Behavior​

Deduplication Key is _kafka_key​

Multiple Deduplication Keys (including _kafka_key plus value fields)​

Troubleshooting​

1. Consumer Group Corruption​

Using OLake UI​

Using OLake CLI​

Prerequisites

Version Prerequisites

Connection Prerequisites

Authentication

1. PLAINTEXT

2. SASL_PLAINTEXT

3. SASL_SSL

Configuration

1. Navigate to the Source Configuration Page

2. Provide Configuration Details

Connection details

Additional configuration

3. Test Connection

1. Create Configuration File

2. Provide Configuration Details

3. Check Source Connection

Kafka General Information

Columns Transferred

Normalization

Stream Configuration

Upsert Behavior

Deduplication Key is `_kafka_key`

Multiple Deduplication Keys (including `_kafka_key` plus value fields)

Troubleshooting

1. Consumer Group Corruption

Using OLake UI

Using OLake CLI