Kafka Source Node

The kafkaSource node in ZephFlow enables you to consume data from Kafka topics, providing a seamless integration point for processing streaming data from Kafka in your data pipelines.

Overview

The Kafka source node connects to a Kafka broker, subscribes to specified topics, and emits messages as records into your ZephFlow pipeline. This allows you to build data processing pipelines that consume and transform data from Kafka in real-time.

Key Features

Kafka Integration: Connect to Kafka brokers to consume messages from topics
Consumer Groups: Leverage Kafka's consumer group functionality for scalable processing
Flexible Encoding: Support for multiple record encoding formats
Customizable Configuration: Fine-tune Kafka consumer settings with custom properties

Parameters

Parameter	Type	Description
`broker`	`String`	Kafka broker address with port (e.g., "localhost:9092")
`topic`	`String`	Name of the Kafka topic to consume from
`groupId`	`String`	Consumer group ID for tracking consumption progress
`encodingType`	`EncodingType`	Format of the messages (e.g., JSON_OBJECT, STRING_LINE, etc.)
`properties`	`Map<String, String>`	Additional Kafka consumer properties (optional)

info

More details about encoding type support can be found here.

Config Object

The full config object for the Kafka source node (KafkaSourceDto.Config):

Field	Type	Required	Default	Description
`broker`	`String`	Yes	—	Kafka broker address with port (e.g., `"localhost:9092"`)
`topic`	`String`	Yes	—	Name of the Kafka topic to consume from
`groupId`	`String`	Yes	—	Consumer group ID for tracking consumption progress
`encodingType`	`EncodingType`	Yes	—	Format of the messages. Supported values: `CSV`, `JSON_OBJECT`, `JSON_ARRAY`, `JSON_OBJECT_LINE`, `STRING_LINE`, `TEXT`, `XML`
`properties`	`Map<String, String>`	No	`null`	Additional Kafka consumer properties
`commitStrategy`	`CommitStrategyType`	No	`BATCH`	Offset commit strategy. Values: `PER_RECORD`, `BATCH`, `NONE`
`commitBatchSize`	`Integer`	No	`1000`	Number of records per commit batch (when `commitStrategy` is `BATCH`)
`commitIntervalMs`	`Long`	No	`5000`	Interval in milliseconds between commits (when `commitStrategy` is `BATCH`)

Default Consumer Settings

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

Property	Default Value	Description
`max.poll.records`	5000	Maximum records returned per poll
`fetch.min.bytes`	1048576 (1MB)	Minimum data to fetch per request
`fetch.max.wait.ms`	1000	Maximum wait time for fetch.min.bytes to be satisfied
`session.timeout.ms`	10000	Consumer session timeout
`max.partition.fetch.bytes`	10485760 (10MB)	Maximum data per partition per fetch
`auto.offset.reset`	earliest	Start from earliest offset when no committed offset exists
`enable.auto.commit`	false	Offsets are committed by ZephFlow, not by Kafka's auto-commit
`allow.auto.create.topics`	false	Subscribing never creates the topic on the broker

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

With allow.auto.create.topics=false, the node checks at startup that the topic exists. If the broker reports the topic as unknown, or the consumer is not authorized to describe it, the node fails with an error instead of polling an empty assignment. Set allow.auto.create.topics to true in properties to skip the check.

tip

The auto.offset.reset=earliest default means new consumer groups will process all available messages from the beginning of the topic. Set it to latest if you only want to process new messages.

Dead Letter Queue (DLQ)

Deserialization failures (e.g., malformed JSON when using JSON_OBJECT) can be routed to a Dead Letter Queue when DLQ is configured at the job level. Failed records are sent to the DLQ topic instead of halting the pipeline.

Best Practices

Consumer Group Strategy

Use meaningful consumer group IDs: Choose descriptive names that identify the application and purpose
Dedicated consumer groups: Use separate consumer groups for different processing pipelines
Consider partitioning: Ensure the number of pipeline instances matches or is a divisor of the number of partitions for balanced consumption

Performance Tuning

Batch size: Adjust max.poll.records based on message size and processing requirements (default: 5000)
Poll interval: Configure fetch.max.wait.ms (default: 1000ms) and fetch.min.bytes (default: 1MB) for optimized throughput
Consumer count: Scale horizontally by running multiple instances of your pipeline

Common Issues and Solutions

Consumer Lag

If your pipeline is falling behind (high consumer lag):

Increase the number of pipeline instances
Optimize downstream processing to reduce latency
Consider increasing fetch sizes and batch processing capacity

Offset Management

Use auto.offset.reset with caution:
- earliest: Process all available messages (good for data completeness)
- latest: Skip to most recent messages (good for real-time monitoring)

DAG Examples

Basic Usage

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "broker-address:9092"     # Kafka broker address
      topic: "topic-name"               # Topic to consume from
      groupId: "consumer-group-id"      # Consumer group ID
      encodingType: "JSON_OBJECT"       # Encoding type of messages
      # properties:                     # Additional Kafka properties (optional)
    outputs:
      - "stdout_sink"

  - id: "stdout_sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"
    outputs: []

Basic JSON Message Consumption

jobContext:
  otherProperties: {}
  metricTags: {}

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

  - id: "add_timestamp"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(processed_timestamp=ts_str_to_epoch($.timestamp, "yyyy-MM-dd'T'HH:mm:ssX")))
    outputs: []

With Custom Kafka Properties

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "logs-topic"
      groupId: "log-processor-group"
      encodingType: "STRING_LINE"
      properties:
        auto.offset.reset: "earliest"
        max.poll.records: "500"
        fetch.max.wait.ms: "500"
    outputs:
      - "print"

  - id: "print"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"
    outputs: []

info

Do not set key.deserializer and value.deserializer in the consumer properties because KafkaSource node will always read raw bytes from kafka and rely on EncodingType configuration for deserialization.

CSV Message Processing

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "metrics-topic"
      groupId: "metrics-processor-group"
      encodingType: "CSV"
    outputs:
      - "to_metric"

  - id: "to_metric"
    commandName: "eval"
    config:
      expression: |
        dict(timestamp=$.0, metric_name=$.1, value=parse_float($.2))
    outputs: []

Real-time Log Processing

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "application-logs"
      groupId: "log-processor-group"
      encodingType: "STRING_LINE"
    outputs:
      - "parse_logs"

  - id: "parse_logs"
    commandName: "parser"
    config:
      targetField: "$"
      extractionConfig:
        type: "grok"
        grokExpression: "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}"
    outputs:
      - "flag_errors"

  - id: "flag_errors"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(is_error=$.level == "ERROR"))
    outputs: []

Enriching Record Data

jobContext:
  otherProperties: {}
  metricTags: {}

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

  - id: "enrich"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(
          event_day=epoch_to_ts_str(ts_str_to_epoch($.timestamp, "yyyy-MM-dd'T'HH:mm:ss"), "yyyy-MM-dd"),
          event_hour=epoch_to_ts_str(ts_str_to_epoch($.timestamp, "yyyy-MM-dd'T'HH:mm:ss"), "HH"),
          user_region=case(str_contains($.user_id, "EU") => "Europe",
            str_contains($.user_id, "US") => "North America",
            _ => "Other")
        ))
    outputs: []

Filtering and Routing Messages

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "transaction-events"
      groupId: "transaction-processor-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "high_value_filter"
      - "regular_filter"

  - id: "high_value_filter"
    commandName: "filter"
    config:
      expression: $.amount > 1000
    outputs:
      - "high_value_sink"

  - id: "high_value_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "high-value-transactions"
      encodingType: "JSON_OBJECT"
    outputs: []

  - id: "regular_filter"
    commandName: "filter"
    config:
      expression: $.amount <= 1000
    outputs:
      - "regular_sink"

  - id: "regular_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka-broker:9092"
      topic: "regular-transactions"
      encodingType: "JSON_OBJECT"
    outputs: []

Performance Tuning

- id: "kafka_source"
  commandName: "kafkasource"
  config:
    broker: "kafka-broker:9092"
    topic: "high-volume-topic"
    groupId: "high-performance-group"
    encodingType: "JSON_OBJECT"
    properties:
      max.poll.records: "1000"
      fetch.max.wait.ms: "500"
      fetch.min.bytes: "65536"

Security Configuration

For secure Kafka clusters, provide the necessary security configurations:

- id: "kafka_source"
  commandName: "kafkasource"
  config:
    broker: "kafka-broker:9092"
    topic: "secure-topic"
    groupId: "secure-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";'

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: []

kafkaSink: Send processed data back to Kafka

Overview​

Key Features​

Parameters​

Config Object​

Default Consumer Settings​

Dead Letter Queue (DLQ)​

Best Practices​

Consumer Group Strategy​

Performance Tuning​

Common Issues and Solutions​

Consumer Lag​

Offset Management​

DAG Examples​

Basic Usage​

Basic JSON Message Consumption​

With Custom Kafka Properties​

CSV Message Processing​

Real-time Log Processing​

Enriching Record Data​

Filtering and Routing Messages​

Performance Tuning​

Security Configuration​

Complete Pipeline​

Related Nodes​

Overview

Key Features

Parameters

Config Object

Default Consumer Settings

Dead Letter Queue (DLQ)

Best Practices

Consumer Group Strategy

Performance Tuning

Common Issues and Solutions

Consumer Lag

Offset Management

DAG Examples

Basic Usage

Basic JSON Message Consumption

With Custom Kafka Properties

CSV Message Processing

Real-time Log Processing

Enriching Record Data

Filtering and Routing Messages

Performance Tuning

Security Configuration

Complete Pipeline

Related Nodes