Kafka Sink Node

The kafkaSink node in ZephFlow enables you to publish processed data to Kafka topics, providing a seamless integration point for sending data from your pipeline to Kafka streams.

Overview

The Kafka sink node connects to a Kafka broker and publishes messages to specified topics. This allows you to build data processing pipelines that transform data and then send the results to Kafka for downstream consumption.

Key Features

Kafka Integration: Connect to Kafka brokers to publish messages to topics
Flexible Encoding: Support for multiple record encoding formats
Partition Key Configuration: Optionally specify a field to use as the partition key
Acknowledged Delivery: Waits for broker acknowledgements by default, with an opt-in fire-and-forget mode
Store and Forward: Optionally buffer records to a local durable queue during a broker outage and replay them on recovery
Managed Credentials: Configure SASL security through a credential reference instead of inline secrets
Customizable Configuration: Fine-tune Kafka producer settings with custom properties

Parameters

Parameter	Type	Description
`broker`	`String`	Kafka broker address with port, comma-separated if multiple (e.g., "kafka-1:9092,kafka-2:9092")
`topic`	`String`	Name of the Kafka topic to publish to
`partitionKeyFieldExpressionStr`	`String`	Expression to determine partition key (null for default partitioning)
`encodingType`	`EncodingType`	Format for serializing messages. Supported types: CSV, JSON_OBJECT, JSON_ARRAY, JSON_OBJECT_LINE. Each record is published as an individual Kafka message.
`properties`	`Map<String, String>`	Additional Kafka producer properties (optional)

The node supports additional configuration fields (delivery mode, store-and-forward, managed SASL credentials) — see Config Object.

info

More details about encoding type support can be found here.

Config Object

The full config object for the Kafka sink node (KafkaSinkDto.Config):

Field	Type	Required	Default	Description
`broker`	`String`	Yes	—	Kafka broker address with port, comma-separated if multiple
`topic`	`String`	Yes	—	Name of the Kafka topic to publish to
`partitionKeyFieldExpressionStr`	`String`	No	`null`	Expression to determine partition key (null for default partitioning)
`encodingType`	`String`	Yes	—	Format for serializing messages. Supported values: `CSV`, `JSON_OBJECT`, `JSON_ARRAY`, `JSON_OBJECT_LINE`
`properties`	`Map<String, String>`	No	`null`	Additional Kafka producer properties
`deliveryMode`	`DeliveryMode`	No	`WAIT_FOR_ACK`	`WAIT_FOR_ACK` waits for broker acknowledgements before the flush returns; `FIRE_AND_FORGET` returns as soon as records are buffered in the producer. A config that omits the field resolves to `WAIT_FOR_ACK`.
`storeAndForwardEnabled`	`boolean`	No	`false`	Buffer records to a local durable queue on a connectivity failure and replay them when the broker is reachable again. Forces acknowledged delivery; cannot be combined with `FIRE_AND_FORGET`.
`localStorePath`	`String`	No	`null`	Directory for the local store-and-forward queue. Defaults to a temp directory keyed by node id.
`localStoreMaxBytes`	`long`	No	`1073741824` (1 GiB)	Hard cap on local store size. Once reached, incoming records are dropped. Must be positive when store-and-forward is enabled.
`forwardRetryIntervalMillis`	`long`	No	`5000`	How often the background forwarder retries draining the local queue during an outage. Must be positive when store-and-forward is enabled.
`credentialId`	`String`	No	`null`	Reference to a stored username/password credential. Required when `securityProtocol` is a `SASL_*` protocol.
`securityProtocol`	`String`	No	`null`	Kafka security protocol (e.g. `SASL_SSL`, `SASL_PLAINTEXT`, `SSL`, `PLAINTEXT`). Setting it switches the node to managed security — see Security Configuration.
`saslMechanism`	`String`	No	`null`	SASL mechanism used with a `SASL_*` protocol. Supported values: `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`.

Default Producer Settings

When no custom properties are provided, the Kafka sink node applies the following defaults optimized for high-throughput production:

Property	Default Value	Description
`batch.size`	65536 (64KB)	Maximum batch size in bytes
`linger.ms`	10	Time to wait for additional messages before sending a batch
`buffer.memory`	67108864 (64MB)	Total memory available to the producer for buffering
`compression.type`	lz4	Compression algorithm for message batches
`acks`	`all` in `WAIT_FOR_ACK` mode, `1` in `FIRE_AND_FORGET` mode	Number of acknowledgments required
`retries`	3	Number of retry attempts on failure
`max.in.flight.requests.per.connection`	5	Maximum unacknowledged requests per connection
`enable.idempotence`	`true` in `WAIT_FOR_ACK` mode (see below)	Deduplicates producer retries

Custom properties passed via the properties parameter will override these defaults.

Automatic Idempotence

In WAIT_FOR_ACK mode the node turns on enable.idempotence for you, but only when the effective producer configuration still supports it — the Kafka producer refuses to start otherwise. Idempotence is left off if you override properties such that any of the following becomes false:

acks is all or -1
retries is greater than 0
max.in.flight.requests.per.connection is 5 or less
enable.idempotence is not already set explicitly in properties

So lowering acks to 1, setting retries=0, or raising max.in.flight.requests.per.connection above 5 silently disables idempotence rather than failing the node.

Store-and-Forward Timeouts

When storeAndForwardEnabled is true, the node additionally bounds the producer's blocking timeouts so an outage surfaces quickly as a buffered batch instead of stalling the pipeline:

Property	Value
`max.block.ms`	5000
`request.timeout.ms`	2000
`delivery.timeout.ms`	5000

These are applied before your properties overrides, so you can still raise them if needed.

Delivery Model

The Kafka sink supports two delivery modes, selected with the deliveryMode config field.

`WAIT_FOR_ACK` (default)

The node sends the batch, flushes the producer, and waits for every broker acknowledgement before the flush returns. This is the default because the source checkpoint advances as soon as the flush returns — returning early would acknowledge records that were never durably written.

A record whose acknowledgement fails becomes an error output, not a silent success
If the broker is unreachable, a single bounded metadata probe fails the batch up front rather than blocking max.block.ms on every individual send
acks=all and, where the configuration allows it, enable.idempotence=true are applied automatically

`FIRE_AND_FORGET`

The node returns as soon as records are in the producer's client-side buffer, and reports delivery through asynchronous callbacks.

Sending a message does not block the pipeline
Delivery errors are tracked asynchronously and reported through counters only — they are not returned to the caller and do not stop the checkpoint from advancing
acks stays at 1 and idempotence is not enabled

Choose FIRE_AND_FORGET only when throughput matters more than knowing a record was written.

Store and Forward

Setting storeAndForwardEnabled to true makes a connectivity failure buffer records to a local durable queue instead of dropping them. A background forwarder drains the queue back to Kafka once the broker is reachable again.

Store-and-forward always uses acknowledged delivery. Combining it with deliveryMode: FIRE_AND_FORGET is rejected at config validation time.
Only transient connectivity failures are buffered. Failures are classified by walking the cause chain for a Kafka RetriableException (such as TimeoutException or NetworkException) or a plain IOException. Anything else is treated as permanent and goes through the normal error path.
localStoreMaxBytes is a hard cap. Once the local store reaches it, incoming records are dropped.
localStorePath defaults to a temp directory keyed by node id.
Store-and-forward is forced off in test mode, since a test run is an ephemeral single shot and a durable queue would leave an orphaned, still-locked on-disk store behind.

The node emits these counters so you can see buffering activity:

Metric	Meaning
`store_forward_buffered_count`	Records written to the local store during an outage
`store_forward_replayed_count`	Records successfully replayed to Kafka after recovery
`store_forward_dropped_count`	Records dropped because the local store was full

Best Practices

Producer Configuration Strategy

Durability vs. Performance: Balance between durability and throughput based on your use case
- For high durability: keep the default WAIT_FOR_ACK mode, which already applies acks=all and idempotence, and increase retries if you need more in-flight retry attempts
- For high throughput: increase batch.size, linger.ms, and use efficient compression. Switching to FIRE_AND_FORGET removes the per-batch ack wait, at the cost of losing delivery failures
- Avoid overriding acks, retries, or max.in.flight.requests.per.connection downward unless you intend to disable idempotence along with them
Partition Key Selection: Choose partition keys that provide even distribution across partitions

Performance Tuning

Batch Size: Increase batch size for higher throughput at the cost of slightly higher latency (default: 64KB)
Compression: Compression is enabled by default (lz4). Switch to snappy or zstd if preferred
Linger Time: Increase linger time to allow more batching for higher throughput (default: 10ms)

Common Issues and Solutions

Message Serialization Errors

Ensure your records match the specified encoding type
When using JSON_OBJECT, make sure your records are valid JSON objects
For other encoding types, ensure your data structure is compatible

Network and Connection Issues

Verify broker addresses and network connectivity
Add appropriate timeouts and retry configurations via custom properties
In the default WAIT_FOR_ACK mode an unreachable broker fails the batch immediately, before any record is sent, instead of blocking on each send
Enable storeAndForwardEnabled if a transient outage should buffer records rather than fail them

Rejected Producer Properties

When securityProtocol is set, the node manages the security properties itself and rejects them in properties. Remove security.protocol, sasl.mechanism, sasl.jaas.config, ssl.keystore.password, ssl.key.password, and ssl.truststore.password from properties and configure them through securityProtocol, saslMechanism, and credentialId instead.

DAG Examples

Basic Usage

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "broker-address:9092"
      topic: "input-topic"
      groupId: "consumer-group-id"
      encodingType: "JSON_OBJECT"
    outputs:
      - "mark_processed"

  - id: "mark_processed"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(processed=true, timestamp=epoch_to_ts_str($.timestamp, "yyyy-MM-dd'T'HH:mm:ss")))
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "broker-address:9092"     # Kafka broker address
      topic: "output-topic"             # Topic to publish to
      encodingType: "JSON_OBJECT"       # Encoding type for messages
      # partitionKeyFieldExpressionStr: # Partition key expression (optional)
      # properties:                     # Additional Kafka properties (optional)
    outputs: []

With Custom Partition Key

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "user-events"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "partitioned-events"
      partitionKeyFieldExpressionStr: "$.user_id"  # Use user_id field as partition key
      encodingType: "JSON_OBJECT"
    outputs: []

With Custom Kafka Properties

- id: "kafka_sink"
  commandName: "kafkasink"
  config:
    broker: "kafka-broker:9092"
    topic: "output-topic"
    encodingType: "JSON_OBJECT"
    properties:
      acks: "all"
      retries: "3"
      batch.size: "16384"
      linger.ms: "5"
  outputs: []

info

Do not set key.serializer and value.serializer in the producer properties because KafkaSink node will always use byte array serializers and rely on EncodingType configuration for serialization.

Record Transformation and Publishing

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "raw-events"
      groupId: "event-processor-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "normalize"

  - id: "normalize"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(
          event_id=$.id,
          event_type=$.type,
          normalized_timestamp=ts_str_to_epoch($.timestamp, "yyyy-MM-dd'T'HH:mm:ss"),
          processed=true
        ))
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "processed-events"
      partitionKeyFieldExpressionStr: "$.event_id"  # Partition by event_id
      encodingType: "JSON_OBJECT"
    outputs: []

Message Routing Based on Content

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "incoming-messages"
      groupId: "router-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "error_filter"
      - "success_filter"

  - id: "error_filter"
    commandName: "filter"
    config:
      expression: $.status == "error"
    outputs:
      - "error_sink"

  - id: "error_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "error-events"
      encodingType: "JSON_OBJECT"
    outputs: []

  - id: "success_filter"
    commandName: "filter"
    config:
      expression: $.status == "success"
    outputs:
      - "success_sink"

  - id: "success_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "success-events"
      encodingType: "JSON_OBJECT"
    outputs: []

Enriching and Aggregating Data

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "user-clicks"
      groupId: "click-enricher-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "enrich"

  - id: "enrich"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(
          session_length=case($.end_time != null and $.start_time != null =>
            ts_str_to_epoch($.end_time, "yyyy-MM-dd'T'HH:mm:ss") -
            ts_str_to_epoch($.start_time, "yyyy-MM-dd'T'HH:mm:ss"),
            _ => 0),
          click_count=size_of($.clicks),
          has_purchase=$.conversion == true
        ))
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "enriched-sessions"
      partitionKeyFieldExpressionStr: "$.session_id"
      encodingType: "JSON_OBJECT"
    outputs: []

High Durability Configuration

The default WAIT_FOR_ACK delivery mode already applies acks=all and enables idempotence, so a plain kafkasink node is durable out of the box. Only raise retries if you want more in-flight retry attempts:

- id: "kafka_sink"
  commandName: "kafkasink"
  config:
    broker: "kafka-broker:9092"
    topic: "output-topic"
    encodingType: "JSON_OBJECT"
    properties:
      retries: "5"
  outputs: []

warning

Setting acks to 1, retries to 0, or max.in.flight.requests.per.connection above 5 in properties silently turns idempotence back off, because the Kafka producer will not start with idempotence enabled under those settings.

Fire-and-Forget Delivery

Set deliveryMode when you want to trade delivery guarantees for throughput:

- id: "kafka_sink"
  commandName: "kafkasink"
  config:
    broker: "kafka-broker:9092"
    topic: "high-volume-topic"
    encodingType: "JSON_OBJECT"
    deliveryMode: "FIRE_AND_FORGET"
  outputs: []

Store and Forward

Buffer records locally during a broker outage and replay them on recovery:

- id: "kafka_sink"
  commandName: "kafkasink"
  config:
    broker: "kafka-broker:9092"
    topic: "output-topic"
    encodingType: "JSON_OBJECT"
    storeAndForwardEnabled: true
    localStorePath: "/var/lib/zephflow/kafka-sink-queue"
    localStoreMaxBytes: 2147483648      # 2 GiB
    forwardRetryIntervalMillis: 5000
  outputs: []

info

storeAndForwardEnabled always waits for broker acks. Setting it together with deliveryMode: FIRE_AND_FORGET fails config validation.

High Throughput Configuration

- id: "kafka_sink"
  commandName: "kafkasink"
  config:
    broker: "kafka-broker:9092"
    topic: "high-volume-topic"
    encodingType: "JSON_OBJECT"
    properties:
      batch.size: "131072"
      linger.ms: "20"
      compression.type: "snappy"
      buffer.memory: "134217728"
  outputs: []

Security Configuration

For secure Kafka clusters you can either supply the raw producer properties yourself, or let the node build them from a stored credential.

Raw properties. Supported when securityProtocol is not set on the config:

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
      properties:
        security.protocol: "SASL_SSL"
        sasl.mechanism: "PLAIN"
        sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="your-username" password="your-password";'
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "secure-topic"
      encodingType: "JSON_OBJECT"
      properties:
        security.protocol: "SASL_SSL"
        sasl.mechanism: "PLAIN"
        sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="your-username" password="your-password";'
    outputs: []

Managed credential. Set securityProtocol, saslMechanism, and credentialId, and the node looks up the username/password and builds the JAAS config for you, so no secret appears in the node config:

jobContext:
  otherProperties:
    my-kafka-credential:
      username: "your-username"
      password: "your-password"

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "secure-topic"
      encodingType: "JSON_OBJECT"
      securityProtocol: "SASL_SSL"
      saslMechanism: "SCRAM-SHA-512"   # PLAIN, SCRAM-SHA-256 or SCRAM-SHA-512
      credentialId: "my-kafka-credential"
    outputs: []

warning

Once securityProtocol is set, the node owns the security properties. Passing security.protocol, sasl.mechanism, sasl.jaas.config, ssl.keystore.password, ssl.key.password, or ssl.truststore.password in properties fails config validation (the check is case-insensitive). credentialId is required for any SASL_* protocol.

Retry Configuration

config:
  properties:
    retries: "5"
    retry.backoff.ms: "500"
    request.timeout.ms: "30000"

Complete Pipeline

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "input-topic"
      groupId: "processor-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "status_filter"

  - id: "status_filter"
    commandName: "filter"
    config:
      expression: $.status != null
    outputs:
      - "add_processed_at"

  - id: "add_processed_at"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(processed_at=epoch_to_ts_str($.timestamp, "yyyy-MM-dd'T'HH:mm:ss")))
    outputs:
      - "check_processed_at"

  - id: "check_processed_at"
    commandName: "assertion"
    config:
      expression: $.processed_at != null
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "output-topic"
      encodingType: "JSON_OBJECT"
    outputs: []

Kafka Source: Read data from Kafka topics

Overview​

Key Features​

Parameters​

Config Object​

Default Producer Settings​

Automatic Idempotence​

Store-and-Forward Timeouts​

Delivery Model​

WAIT_FOR_ACK (default)​

FIRE_AND_FORGET​

Store and Forward​

Best Practices​

Producer Configuration Strategy​

Performance Tuning​

Common Issues and Solutions​

Message Serialization Errors​

Network and Connection Issues​

Rejected Producer Properties​

DAG Examples​

Basic Usage​

With Custom Partition Key​

With Custom Kafka Properties​

Record Transformation and Publishing​

Message Routing Based on Content​

Enriching and Aggregating Data​

High Durability Configuration​

Fire-and-Forget Delivery​

Store and Forward​

High Throughput Configuration​

Security Configuration​

Retry Configuration​

Complete Pipeline​

Related Nodes​

Overview

Key Features

Parameters

Config Object

Default Producer Settings

Automatic Idempotence

Store-and-Forward Timeouts

Delivery Model

`WAIT_FOR_ACK` (default)

`FIRE_AND_FORGET`

Store and Forward

Best Practices

Producer Configuration Strategy

Performance Tuning

Common Issues and Solutions

Message Serialization Errors

Network and Connection Issues

Rejected Producer Properties

DAG Examples

Basic Usage

With Custom Partition Key

With Custom Kafka Properties

Record Transformation and Publishing

Message Routing Based on Content

Enriching and Aggregating Data

High Durability Configuration

Fire-and-Forget Delivery

Store and Forward

High Throughput Configuration

Security Configuration

Retry Configuration

Complete Pipeline

Related Nodes