> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/connectors/message-queues/kafka.md).
# Kafka
**Source** | **Sink**
The Kafka connector lets you read from and write to Apache Kafka topics. It works with any Kafka-compatible broker - self-hosted Apache Kafka, Amazon MSK, Redpanda, or any other system that speaks the Kafka protocol. You can use it as a source (consume messages into a pipeline), as a sink (produce messages out of a pipeline), or both in the same deployment.
The connector supports SASL authentication (SCRAM-SHA-256, SCRAM-SHA-512, PLAIN), AWS MSK IAM authentication, and optional integration with Confluent Schema Registry for Avro and Protobuf schemas. For Confluent Cloud specifically, see the [Confluent connector](/ingestion-engine/connectors/message-queues/confluent.md) which simplifies the authentication setup.
***
## Prerequisites
### Kafka Cluster Access
* A running Kafka cluster (Apache Kafka, Amazon MSK, Redpanda, or any Kafka-protocol-compatible broker)
* Topics created - e6 Ingestion Engine consumes from and produces to existing topics, it does not create them
* Credentials matching your cluster's auth method (SASL, MSK IAM, or none for local dev)
* Schema Registry endpoint and credentials (only if using Avro or Protobuf with Confluent Schema Registry)
### ACLs / Permissions
If your cluster uses ACLs, the e6 Ingestion Engine service account needs:
**Source (consuming from a topic):**
* `READ` + `DESCRIBE` on the topic
* `READ` on the consumer group (see [Consumer Group](#consumer-group) for naming conventions)
* `DESCRIBE` on the cluster
**Sink (producing to a topic) - add to the above:**
* `WRITE` + `DESCRIBE` on the sink topic
* `IDEMPOTENT_WRITE` on the cluster
* `WRITE` + `DESCRIBE` on the transactional ID (only for `exactly_once` mode - the transactional ID format is `laminar-id-{job_id}-{operator_id}-{topic}-{subtask_index}-{transaction_index}`)
### Network Access
e6 Ingestion Engine initiates outbound connections to the Kafka brokers. No inbound access from the broker side is required.
{% tabs %}
{% tab title="Azure" %}
* **Kafka brokers**: outbound TCP to the broker port (typically `9092` for SASL\_SSL, `9093` for SSL) from the e6 Ingestion Engine worker pods
* **Schema Registry**: outbound HTTPS port `443` if using Confluent Schema Registry
* For **Azure Event Hubs** (Kafka-compatible): outbound TCP port `9093` (SASL\_SSL)
* For **Amazon MSK** accessed cross-cloud: ensure the broker endpoints are reachable from the Azure VNet - this may require VPN or peering. If MSK uses IAM auth, the worker pods need AWS credentials (typically via environment variables or mounted secrets).
* If using IP-restricted brokers, allowlist the NAT Gateway public IPs of the e6data compute plane
{% endtab %}
{% tab title="AWS" %}
* **Kafka brokers**: outbound TCP to the broker port (typically `9092`) from the e6 Ingestion Engine worker pods
* **Schema Registry**: outbound HTTPS port `443` if using Confluent Schema Registry
* For **Amazon MSK** in the same VPC/account: ensure the security group on the MSK cluster allows inbound from the e6 Ingestion Engine worker security group on port `9092` (or `9098` for IAM)
* For **Amazon MSK** with IAM auth: the worker pods' IAM role needs `kafka-cluster:Connect`, `kafka-cluster:ReadData`, `kafka-cluster:DescribeTopic`, and `kafka-cluster:DescribeGroup` permissions. Add `kafka-cluster:WriteData` and `kafka-cluster:WriteDataIdempotently` for sinks.
* For MSK Serverless or PrivateLink-enabled clusters, work with your e6data SE to configure VPC endpoint connectivity
{% endtab %}
{% tab title="GCP" %}
* **Kafka brokers**: outbound TCP to the broker port (typically `9092`) from the e6 Ingestion Engine worker pods
* **Schema Registry**: outbound HTTPS port `443` if using Confluent Schema Registry
* If using a self-hosted Kafka cluster in GCP, ensure firewall rules allow inbound from the e6 Ingestion Engine worker node IPs or NAT Gateway IPs on the broker port
* For cross-cloud access (e.g., AWS MSK from GKE), ensure network connectivity via VPN, peering, or public endpoints with IP filtering
{% endtab %}
{% endtabs %}
***
## Connection
The connection config tells e6 Ingestion Engine how to reach your Kafka cluster. It holds broker addresses and credentials, and is shared across all sources and sinks that use the same cluster.
```yaml
bootstrap_servers: broker1:9092,broker2:9092
authentication:
type: none
```
* **`bootstrap_servers`** - Comma-separated list of Kafka brokers. e6 Ingestion Engine uses these for initial cluster discovery - it contacts these brokers to learn the full cluster topology, so you don't need to list every broker.
* **`authentication`** - How to authenticate with the cluster. See below for options.
### Authentication
Choose the method that matches your cluster setup.
**No authentication** - for local development or clusters without auth enabled:
```yaml
authentication:
type: none
```
**SASL** - the most common method for production Kafka clusters. Supports SCRAM-SHA-256, SCRAM-SHA-512, and PLAIN mechanisms over either SSL or plaintext transport:
```yaml
authentication:
type: sasl
sasl_config:
protocol: SASL_SSL
mechanism: SCRAM-SHA-256
username: your-username
password: your-password
```
* **`protocol`** - Transport security. `SASL_SSL` (encrypted, recommended for production) or `SASL_PLAINTEXT` (unencrypted, only for trusted networks). Maps directly to the rdkafka `security.protocol` setting.
* **`mechanism`** - SASL mechanism. `SCRAM-SHA-256`, `SCRAM-SHA-512`, or `PLAIN`. Must match your broker's configured SASL mechanisms. Maps to rdkafka `sasl.mechanism`.
* **`username`** - SASL username. Supports environment variable substitution - use `${ENV_VAR}` syntax to avoid storing credentials in config files.
* **`password`** - SASL password. Supports `${ENV_VAR}` substitution.
**AWS MSK IAM** - for Amazon MSK clusters using IAM authentication. e6 Ingestion Engine uses the OAUTHBEARER mechanism under the hood, automatically generating and refreshing IAM auth tokens. The worker pods must have an IAM role with `kafka-cluster:*` permissions:
```yaml
authentication:
type: aws_msk_iam
aws_msk_iam_config:
region: us-east-1
```
* **`region`** - AWS region where the MSK cluster is deployed. Used to generate the IAM auth token.
### Schema Registry
If your Kafka topics use Avro or Protobuf schemas managed by a Confluent Schema Registry, add schema registry config to the connection. This is optional - skip it if your topics use JSON or raw bytes.
```yaml
schema_registry:
endpoint: https://schema-registry.example.com
api_key: your-api-key
api_secret: your-api-secret
```
* **`endpoint`** (required) - Schema Registry URL. Used to initialize the schema registry client that fetches Avro and Protobuf schemas for deserialization.
* **`api_key`** (optional) - API key for schema registry authentication. Supports `${ENV_VAR}` substitution.
* **`api_secret`** (optional) - API secret. Supports `${ENV_VAR}` substitution.
When schema registry is configured, the connector automatically deserializes messages using the schema associated with the topic. The subject name defaults to `{topic}-value` but can be overridden per table with the `value_subject` property.
### Extra Broker Properties
The `connection_properties` field lets you pass arbitrary [librdkafka configuration properties](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) to the underlying Kafka client. These are applied to every consumer and producer that uses this connection.
```yaml
connection_properties:
socket.timeout.ms: "30000"
ssl.ca.location: /etc/ssl/certs/ca-certificates.crt
```
e6 Ingestion Engine sets sensible defaults for common rdkafka properties. You can override any of them via `connection_properties` or per-table `client_configs`:
* `socket.timeout.ms` = `120000` - Socket timeout (2 minutes). Increase for high-latency connections.
* `request.timeout.ms` = `120000` - Request timeout (2 minutes).
* `session.timeout.ms` = `60000` - Consumer session timeout (1 minute). If no heartbeat is received within this period, the broker considers the consumer dead.
* `fetch.wait.max.ms` = `30000` - Maximum time the broker waits for enough data before returning a fetch response.
* `reconnect.backoff.ms` = `1000` - Initial reconnect backoff (1 second).
* `reconnect.backoff.max.ms` = `10000` - Maximum reconnect backoff (10 seconds).
* `socket.keepalive.enable` = `true` - TCP keepalive for long-lived connections.
* `connections.max.idle.ms` = `540000` - Close idle connections after 9 minutes.
* `enable.auto.commit` = `false` - Always disabled. e6 Ingestion Engine manages offsets through its own checkpoint system, not Kafka consumer group commits.
* `enable.partition.eof` = `false` - Disabled. The consumer does not emit EOF signals when reaching the end of a partition.
***
## Reading from Kafka
To consume messages from a topic, configure a source with the topic name, a schema describing the message structure, and an offset mode.
```yaml
topic: user-events
type: source
source:
offset: latest
schema:
format:
json: {}
fields:
- field_name: user_id
field_type:
type:
primitive: Int64
nullable: false
- field_name: event_type
field_type:
type:
primitive: Utf8
nullable: false
```
### Topic and Schema
The `topic` field specifies the Kafka topic to consume from. The `schema` block defines the expected structure of the messages - e6 Ingestion Engine uses this to deserialize incoming records into typed Arrow columns. See the [Schema Reference](/ingestion-engine/get-started/schema.md) for field type details.
The `format` field inside `schema` determines how messages are deserialized:
* **`json: {}`** - JSON messages (most common)
* **`avro: {}`** - Avro with Schema Registry
* **`protobuf: {}`** - Protobuf with Schema Registry
* **`raw_bytes: {}`** - Pass through as raw bytes
### Metadata Fields
Every message consumed from Kafka carries metadata that e6 Ingestion Engine makes available as additional fields in your pipeline. These can be referenced in SQL queries and transformations:
* **`offset_id`** (Int64) - Kafka offset of the message within its partition
* **`partition`** (Int32) - Partition number the message was consumed from
* **`topic`** (Utf8) - Topic name (useful when consuming from multiple topics)
* **`timestamp`** (Int64) - Kafka message timestamp (milliseconds since epoch)
* **`key`** (Binary) - Raw message key bytes
### Where to Start Reading
The `offset` field determines where e6 Ingestion Engine begins consuming when a pipeline starts for the first time:
* **`latest`** - Start from the newest messages. Skip everything already in the topic. Use this when you only care about new data going forward.
* **`earliest`** - Start from the oldest available messages. Replay the entire topic from the beginning. Use this for backfills or when you need complete history.
* **`group`** - Resume from the last committed offset for the consumer group. Use this when you want Kafka-managed offset tracking in addition to e6 Ingestion Engine's checkpoints.
**Important:** The `offset` setting only applies on the very first start of a pipeline, or when there is no checkpoint to restore from. After that, e6 Ingestion Engine tracks offsets through its own checkpoint system. On recovery from a failure, it always resumes from the last successful checkpoint - the `offset` setting is ignored. If new partitions appear that were not known at checkpoint time, those new partitions start from the configured `offset`.
### Read Mode
By default, the consumer sees all messages including those from uncommitted Kafka transactions. If your producers use Kafka transactions and you need transactional isolation, set the read mode:
* **`read_uncommitted`** (default) - See all messages immediately. Higher throughput.
* **`read_committed`** - Only see messages from committed transactions. Sets rdkafka `isolation.level` to `read_committed`. Adds latency equal to the transaction timeout since the consumer must wait for transactions to complete.
```yaml
source:
offset: latest
read_mode: read_committed
```
### Consumer Group
e6 Ingestion Engine auto-generates a consumer group ID for each pipeline in the format `laminar-{job_id}-{operator_id}-consumer`. This ensures each pipeline instance gets its own consumer group. You can override this:
* **`group_id`** - Explicit consumer group ID. Takes precedence over auto-generation. Use this when you need a predictable group name for monitoring dashboards, Kafka ACLs, or offset inspection tools.
* **`group_id_prefix`** - Prefix for the auto-generated group ID. The resulting ID becomes `{prefix}-laminar-{job_id}-{operator_id}`. Useful for distinguishing environments (e.g., `prod-`, `staging-`). Ignored if `group_id` is set.
Note that e6 Ingestion Engine does not rely on Kafka consumer group offsets for recovery - it uses its own checkpoint-based offset tracking. However, it does perform an async offset commit to Kafka after each checkpoint, which makes consumer lag visible in Kafka monitoring tools.
### Schema Registry Subject
If the connection has schema registry configured, the source uses the subject `{topic}-value` by default to look up the schema for deserialization. Override this if your subject naming convention differs:
```yaml
value_subject: my-custom-subject-name
```
### Extra Consumer Properties
The `client_configs` field lets you pass additional Kafka consumer properties for this specific source. These are merged with `connection_properties` from the connection profile - if the same key appears in both, the table-level value wins (and a warning is logged):
```yaml
client_configs:
fetch.min.bytes: "1048576"
fetch.wait.max.ms: "500"
```
***
## Writing to Kafka
To produce messages to a topic, configure a sink with the topic name, a delivery guarantee, and a schema describing the output messages.
```yaml
topic: processed-events
type: sink
sink:
commit_mode: at_least_once
schema:
format:
json: {}
fields:
- field_name: user_id
field_type:
type:
primitive: Int64
nullable: false
- field_name: event_type
field_type:
type:
primitive: Utf8
nullable: false
```
### Delivery Guarantees
The `commit_mode` field controls the delivery guarantee for messages written to Kafka:
* **`at_least_once`** - Messages are guaranteed to be delivered but may be duplicated if a failure occurs and the pipeline restarts. Uses a standard Kafka producer with no transaction overhead. Best for most use cases - throughput is higher and duplicates are rare in practice. Downstream consumers should be idempotent or tolerant of occasional duplicates.
* **`exactly_once`** - No duplicates. Enables Kafka idempotent producer (`enable.idempotence=true`) and uses Kafka transactions with a unique `transactional.id` per subtask. Each checkpoint initiates a transaction commit. Use when duplicates are unacceptable - financial transactions, billing events, audit logs. Requires Kafka 0.11+.
In `exactly_once` mode, e6 Ingestion Engine generates a transactional ID in the format `laminar-id-{job_id}-{operator_id}-{topic}-{subtask_index}-{transaction_index}`. The transaction index increments with each checkpoint, ensuring unique transaction fencing across restarts.
### Message Keys and Timestamps
By default, messages are written without a key and use the current time as the Kafka timestamp. You can override both:
* **`key_field`** - Name of a field in the schema to use as the Kafka message key. The field **must be of type `Utf8`** - if the field doesn't exist or is a different type, a warning is logged and the key is not set. Messages with the same key land in the same partition, preserving ordering for that key.
* **`timestamp_field`** - Name of a field in the schema to use as the Kafka message timestamp. The field **must be of type `TimestampNanosecond`** - if the field doesn't exist or is a different type, a warning is logged and the default timestamp is used instead.
Setting a `key_field` is important when downstream consumers rely on per-key ordering - for example, when writing change events where updates to the same entity must be consumed in order.
```yaml
sink:
commit_mode: exactly_once
key_field: user_id
timestamp_field: event_time
```
### Schema Registry Subject
Like the source, the sink defaults to `{topic}-value` for the schema registry subject. Override with `value_subject` if needed.
### Extra Producer Properties
The `client_configs` field lets you pass additional Kafka producer properties for this specific sink. Same merge behavior as the source - table-level values override connection-level `connection_properties`:
```yaml
client_configs:
linger.ms: "50"
compression.type: "lz4"
```
***
## Complete Example
A full configuration reading JSON events from Kafka and writing processed results back to a different topic.
**Connection:**
```yaml
bootstrap_servers: kafka-1.prod:9092,kafka-2.prod:9092,kafka-3.prod:9092
authentication:
type: sasl
sasl_config:
protocol: SASL_SSL
mechanism: SCRAM-SHA-256
username: "${KAFKA_USERNAME}"
password: "${KAFKA_PASSWORD}"
schema_registry:
endpoint: https://schema-registry.prod:8081
```
**Source** - read user events, starting from latest, with transactional isolation:
```yaml
topic: user-events
type: source
source:
offset: latest
read_mode: read_committed
group_id_prefix: prod-
schema:
format:
json: {}
fields:
- field_name: user_id
field_type:
type:
primitive: Int64
nullable: false
- field_name: event_type
field_type:
type:
primitive: Utf8
nullable: false
- field_name: timestamp
field_type:
type:
primitive: TimestampNanosecond
nullable: false
- field_name: payload
field_type:
type:
primitive: Utf8
nullable: true
```
**Sink** - write processed events with exactly-once delivery, keyed by user:
```yaml
topic: processed-events
type: sink
sink:
commit_mode: exactly_once
key_field: user_id
timestamp_field: timestamp
schema:
format:
json: {}
fields:
- field_name: user_id
field_type:
type:
primitive: Utf8
nullable: false
- field_name: event_type
field_type:
type:
primitive: Utf8
nullable: false
- field_name: timestamp
field_type:
type:
primitive: TimestampNanosecond
nullable: false
```
***
## JSON Schema Reference
Connection Schema
```json
{
"type": "object",
"properties": {
"bootstrap_servers": {
"type": "string",
"description": "Comma-separated list of Kafka servers to connect to"
},
"authentication": {
"oneOf": [
{
"type": "object",
"title": "None",
"required": ["type"],
"properties": {
"type": {"const": "none"}
}
},
{
"type": "object",
"title": "SASL",
"required": ["type", "sasl_config"],
"properties": {
"type": {"const": "sasl"},
"sasl_config": {
"type": "object",
"required": ["protocol", "mechanism", "username", "password"],
"properties": {
"protocol": {"type": "string"},
"mechanism": {"type": "string"},
"username": {"type": "string"},
"password": {"type": "string"}
}
}
}
},
{
"type": "object",
"title": "AWS_MSK_IAM",
"required": ["type", "aws_msk_iam_config"],
"properties": {
"type": {"const": "aws_msk_iam"},
"aws_msk_iam_config": {
"type": "object",
"required": ["region"],
"properties": {
"region": {"type": "string"}
}
}
}
}
]
},
"schema_registry": {
"oneOf": [
{"type": "object", "title": "None"},
{
"type": "object",
"title": "Confluent Schema Registry",
"required": ["endpoint"],
"properties": {
"endpoint": {"type": "string"},
"api_key": {"type": "string"},
"api_secret": {"type": "string"}
}
}
]
},
"connection_properties": {"type": "object"}
},
"required": ["bootstrap_servers", "authentication"]
}
```
Table Schema
```json
{
"type": "object",
"properties": {
"topic": {"type": "string"},
"type": {"type": "string", "enum": ["source", "sink"]},
"source": {
"type": "object",
"description": "Source-specific config (when type=source)",
"properties": {
"offset": {"type": "string", "enum": ["latest", "earliest", "group"]},
"read_mode": {"type": "string", "enum": ["read_uncommitted", "read_committed"]},
"group_id": {"type": "string"},
"group_id_prefix": {"type": "string"}
},
"required": ["offset"]
},
"sink": {
"type": "object",
"description": "Sink-specific config (when type=sink)",
"properties": {
"commit_mode": {"type": "string", "enum": ["at_least_once", "exactly_once"]},
"key_field": {"type": "string"},
"timestamp_field": {"type": "string"}
},
"required": ["commit_mode"]
},
"client_configs": {"type": "object"},
"value_subject": {"type": "string"}
},
"required": ["topic", "type"]
}
```
---
# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.
## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:
```
GET https://docs.e6data.com/ingestion-engine/connectors/message-queues/kafka.md?ask=&goal=
```
`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.