Kafka sink

The Kafka sink publishes batches to a Kafka topic using rdkafka.

When to use Kafka

Kafka excels as the backbone for event-driven architectures where durability, ordering, and replay capabilities are critical.

Real-world applications

Pros and cons

Configuration

sinks:
  - type: kafka
    config:
      id: orders-kafka
      brokers: ${KAFKA_BROKERS}
      topic: orders
      required: true
      exactly_once: false
      client_conf:
        message.timeout.ms: "5000"
        acks: "all"

Transactional atomic-batch delivery (exactly_once: true)

When exactly_once: true, the Kafka sink uses a transactional producer. Each batch is wrapped in a Kafka transaction (begin_transaction / commit_transaction), so a read_committed consumer sees either all events in the batch or none - never a partial batch.

This is not end-to-end exactly-once. DeltaForge advances its own checkpoint only after the Kafka transaction commits, so a crash between the commit and the checkpoint persist replays that batch on restart as a new transaction, and a read_committed consumer sees it twice. Delivery is at-least-once across a restart; dedup on the event id (stable across replays) to reach exactly-once end to end. The config field is named exactly_once for historical reasons; read it as “transactional producer.”

sinks:
  - type: kafka
    config:
      id: orders-kafka
      brokers: ${KAFKA_BROKERS}
      topic: orders
      exactly_once: true

How it works

1. DeltaForge assigns a stable transactional.id per pipeline-sink pair (deltaforge-{pipeline}-{sink_id})
2. On startup, init_transactions() registers with the broker and fences any zombie producer from a previous instance
3. Each batch: begin_transaction() → produce messages → commit_transaction()
4. If the commit fails before completing, the transaction is aborted and the batch is re-attempted from the same checkpoint (bounded in-sink retry); a persistent failure holds the checkpoint and recovers by source replay on restart
5. Consumers using isolation.level=read_committed only see committed batches; a batch replayed after a restart appears as a new committed transaction (duplicate), so consumers still dedup on event id

Requirements

• Kafka 2.5+ (transaction support)
• isolation.level=read_committed on consumers (otherwise they see uncommitted messages)
• Only one DeltaForge instance per pipeline at a time (the broker will fence duplicates)

Transactional overrides

When exactly_once: true, DeltaForge automatically configures:

You do not need to set these — they are applied automatically as defaults. Your client_conf is merged last, so any value you set there takes precedence (override at your own risk; these defaults are chosen for correct transactional behavior).

Fatal errors

If the broker fences the producer (another instance started with the same transactional.id), DeltaForge treats this as a fatal error and stops the pipeline. This is not retryable — resolve the duplicate instance before restarting.

Performance impact

Transactions add ~1-3ms overhead per batch for the two-phase commit. With properly sized batches (max_events=16000, max_bytes=16MB), throughput impact is ~7-11%. See the Performance guide for tuning details and benchmark results.

Avro encoding threads

An Avro sink encodes each batch in full before it enqueues any record, so a batch whose schema cannot be resolved publishes nothing. With encode_workers: N (N > 1) the batch is split into N contiguous slices, encoded on N threads, and enqueued in its original order: the records, their order, the batch’s failure behavior and its acknowledgement are the same as with one thread. A record whose schema is not registered yet (a table’s first batch, a new schema version) is encoded on the delivery task, which registers the schema once. A record that cannot be encoded is handled as with one thread. Only one batch is encoded at a time, so memory holds one batch’s encoded records whatever the thread count; the threads exist only while a batch is encoded. JSON sinks ignore the setting.

One thread (the default) keeps encoding on the delivery task and uses the least CPU. More threads are a capacity option for saturated Avro streams on hosts with free cores. Measured on one host (a 15 W laptop CPU, MySQL to Kafka, saturated backlog, five interleaved repetitions):

The extra CPU per row with more threads (1.45 to 2.44 CPU seconds per 100k narrow rows at four) is mostly lower clock speed on a power-limited CPU running more cores: the instructions per row do not grow. Records per Kafka request stay the same (about 970-980 narrow, 660 wide).

Linger defaults by format

client_conf overrides both.

Recommended client_conf settings

Consuming events

Kafka CLI

# Consume from beginning
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --from-beginning

# Consume with consumer group
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --group deltaforge-consumers

Go consumer example

config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
config.Consumer.Offsets.Initial = sarama.OffsetOldest

group, _ := sarama.NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)

for {
    err := group.Consume(ctx, []string{"orders"}, &handler{})
    if err != nil {
        log.Printf("Consumer error: %v", err)
    }
}

type handler struct{}

func (h *handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var event Event
        json.Unmarshal(msg.Value, &event)
        process(event)
        session.MarkMessage(msg, "")
    }
    return nil
}

Rust consumer example

#![allow(unused)]
fn main() {
use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::Message;

let consumer: StreamConsumer = ClientConfig::new()
    .set("bootstrap.servers", "localhost:9092")
    .set("group.id", "my-group")
    .set("auto.offset.reset", "earliest")
    .create()?;

consumer.subscribe(&["orders"])?;

loop {
    match consumer.recv().await {
        Ok(msg) => {
            let payload = msg.payload_view::<str>().unwrap()?;
            let event: Event = serde_json::from_str(payload)?;
            process(event);
            consumer.commit_message(&msg, CommitMode::Async)?;
        }
        Err(e) => eprintln!("Kafka error: {}", e),
    }
}
}

Python consumer example

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value
    process(event)
    consumer.commit()

Failure modes

In the table below, “retries” means a bounded in-sink retry (exponential backoff, finite attempts) inside a single send_batch call. DeltaForge does not run an unbounded pipeline-level retry loop: once in-sink attempts are exhausted on a required sink, the checkpoint is held and the batch is recovered by source replay on the next restart. “Blocks checkpoint” means the source backpressures until the sink recovers.

Failure scenarios and data guarantees

Broker failure during batch delivery

1. DeltaForge sends batch of 100 events
2. 50 events delivered, broker crashes
3. rdkafka detects failure, retries remaining 50
4. If idempotence enabled: no duplicates
5. If not: possible duplicates of events near failure point
6. Checkpoint only saved after ALL events acknowledged

DeltaForge crash after Kafka ack, before checkpoint

1. Batch delivered to Kafka successfully
2. DeltaForge crashes before saving checkpoint
3. On restart: replays from last checkpoint
4. Without exactly_once: duplicate events in Kafka (at-least-once); consumer must handle idempotently
5. With exactly_once: replayed batch gets a new transaction; consumers with read_committed see duplicates but each batch is atomic

Monitoring recommendations

DeltaForge exposes these metrics for Kafka sink monitoring:

# DeltaForge sink metrics (exposed at /metrics on port 9000)
deltaforge_sink_events_total{pipeline,sink}      # Events delivered
deltaforge_sink_batch_total{pipeline,sink}       # Batches delivered
deltaforge_sink_latency_seconds{pipeline,sink}   # Delivery latency histogram
deltaforge_stage_latency_seconds{pipeline,stage="sink"}  # Stage timing

For deeper Kafka broker visibility, monitor your Kafka cluster directly:

• Broker metrics via JMX or Kafka’s built-in metrics
• Consumer lag via kafka-consumer-groups.sh
• Topic throughput via broker dashboards

Note: Internal rdkafka producer statistics (message queues, broker RTT, etc.) are not currently exposed by DeltaForge. This is a potential future enhancement.

Notes

• Combine Kafka with other sinks to fan out data; use commit policy to control checkpoint behavior
• For transactional atomic-batch delivery, set exactly_once: true and ensure your Kafka cluster supports transactions (2.5+); this is not end-to-end exactly-once, so consumers still dedup on event id across restarts
• With exactly_once: false (default), idempotent production is still enabled — duplicates are prevented during retries but not across DeltaForge restarts
• Adjust client_conf for durability (acks=all) or performance based on your requirements
• Consider partitioning strategy for ordering guarantees within partitions