Transactions and exactly-once processing

Consume-transform-produce pipelines need exactly-once results, which at-least-once plus idempotence alone cannot give.

Idempotence prevents duplicates from retries within one producer session. Transactions go further: a producer can write to several partitions and topics and commit consumer offsets atomically. Either all of it becomes visible or none does.

The flow for a consume-transform-produce loop:

configure the producer with a stable, unique transactional.id and call init_transactions() once (this also fences off zombie instances using the same id)
for each batch: begin_transaction(), produce outputs, send_offsets_to_transaction(positions, group_metadata), then commit_transaction()
on an error, abort_transaction(); the outputs are discarded and the offsets are not committed, so the input is reprocessed

Consumers of the output topic must use isolation.level=read_committed to skip aborted records and wait for commits. This is the default in librdkafka but not in the Java client (read_uncommitted).

Exactly-once applies within Kafka. Side effects outside Kafka (sending email, writing to a database) are not covered; for those use idempotent writes or store offsets with the external data.

Transactions add latency (commit markers, coordination) and throughput cost if batches are tiny; commit every few hundred records or milliseconds. Kafka Streams uses all of this internally with processing.guarantee=exactly_once_v2.

Example

from confluent_kafka import Consumer, Producer

consumer = Consumer({"bootstrap.servers": "localhost:9092", "group.id": "enricher",
                     "enable.auto.commit": False, "isolation.level": "read_committed"})
producer = Producer({"bootstrap.servers": "localhost:9092",
                     "transactional.id": "enricher-1"})   # stable per instance
consumer.subscribe(["orders"])
producer.init_transactions()

while True:
    batch = consumer.consume(num_messages=200, timeout=1.0)
    if not batch:
        continue
    producer.begin_transaction()
    try:
        for msg in batch:
            producer.produce("orders-enriched", key=msg.key(), value=msg.value().upper())
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()), consumer.consumer_group_metadata())
        producer.commit_transaction()          # outputs + offsets, atomically
    except Exception:
        producer.abort_transaction()           # nothing visible, input re-read
        raise

Remember

Transactions atomically cover produced records and consumed offsets
A stable transactional.id fences zombie instances
Downstream consumers need isolation.level=read_committed
Exactly-once stops at Kafka's boundary; external side effects need idempotence

Watch out for

Assuming a transactional pipeline makes an external API call exactly-once.

Check yourself

Why must output consumers use read_committed in a transactional pipeline?

Official docs: kafka.apache.org/43/design/design/#using-transactions

Example

Remember

Watch out for

Check yourself

Builds on

Unlocks