Module 02: Delivery Semantics and Exactly-Once

Why this matters

"Exactly-once" is the most misunderstood phrase in streaming. It's often dismissed as impossible (true, for delivery over an unreliable network) and just as often assumed to be a checkbox (false). Kafka does provide genuine exactly-once processing — but only within a specific boundary, and only if you use the right API. Outside that boundary, the honest answer is at-least-once plus idempotent handlers, and knowing which situation you're in is the actual skill.

Getting this wrong is expensive in a way that's specific to money: a payment consumer that reprocesses a batch after a rebalance charges customers twice.

Concepts

The three semantics, and where they come from

The first two are decided by one line of ordering:

# AT-MOST-ONCE — do not do this unless loss is genuinely acceptable
msg = consumer.poll(1.0)
consumer.commit(msg)          # committed first
process(msg)                  # crash here -> record never processed, never retried

# AT-LEAST-ONCE — the correct default
msg = consumer.poll(1.0)
process(msg)                  # crash here -> record redelivered to the next owner
consumer.commit(msg)          # committed only after success

Auto-commit is neither, reliably

{"enable.auto.commit": True, "auto.commit.interval.ms": 5000}

Auto-commit fires on a timer, during poll(), for everything returned by the previous poll — whether or not you finished processing it. So it's at-least-once when your batch completes faster than the interval, and at-most-once when it doesn't. You don't get to choose which, and it varies with load.

Auto-commit is fine for genuinely idempotent, loss-tolerant work (metrics, logs, cache warming). For anything else, turn it off:

{"enable.auto.commit": False}

At-least-once, done properly

from confluent_kafka import Consumer, KafkaError

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "payments",
    "enable.auto.commit": False,
    "auto.offset.reset": "earliest",
    "max.poll.records": 20,           # module 01: keep batches small
    "partition.assignment.strategy": "cooperative-sticky",
})
consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise KafkaException(msg.error())

        process(msg)                              # must be idempotent
        consumer.commit(msg, asynchronous=False)  # synchronous: know it landed
finally:
    consumer.close()                              # commits + leaves group cleanly

Committing synchronously per record is the safest and slowest option. Committing per batch is the usual compromise — you narrow the reprocessing window to one batch rather than one record, which is why small max.poll.records helps twice.

The offset you commit is offset + 1

This trips people up constantly. A committed offset means "the next record I want", not "the last record I processed":

from confluent_kafka import TopicPartition

# commit position AFTER msg
consumer.commit(offsets=[TopicPartition(msg.topic(), msg.partition(), msg.offset() + 1)])

consumer.commit(message=msg) does the +1 for you. Doing it manually and forgetting the +1 means every restart reprocesses one record per partition — a subtle, low-grade duplicate source that's hard to spot.

Idempotent processing beats clever delivery

For most systems, the right answer is at-least-once delivery + an idempotent handler, not transactions. Track 10's idempotency-key pattern applies directly:

def process(msg):
    event = json.loads(msg.value())
    key = event["event_id"]                      # producer-assigned, stable
    with db.transaction():
        # UNIQUE(event_id) makes the second attempt a no-op
        inserted = db.execute(
            "INSERT INTO processed_events (event_id) VALUES (%s) "
            "ON CONFLICT (event_id) DO NOTHING RETURNING event_id", (key,))
        if not inserted:
            return                               # already handled; skip silently
        apply_business_effect(event)

This works regardless of Kafka's guarantees, survives consumer restarts and rebalances, and — crucially — extends to side effects Kafka transactions can't cover at all, like charging a card or sending an email.

Kafka transactions: exactly-once within Kafka

Transactions solve one specific shape: consume from Kafka, process, produce to Kafka — atomically, including the offset commit.

from confluent_kafka import Producer, Consumer, TopicPartition

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "payment-processor-1",   # MUST be stable per instance
    "enable.idempotence": True,                  # implied, but be explicit
})
producer.init_transactions()

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "payments",
    "enable.auto.commit": False,                 # transaction owns the offsets
    "isolation.level": "read_committed",         # don't read aborted data
})
consumer.subscribe(["orders"])

while True:
    msgs = consumer.consume(num_messages=100, timeout=1.0)
    if not msgs:
        continue

    producer.begin_transaction()
    try:
        for msg in msgs:
            if msg.error():
                continue
            producer.produce("payments", key=msg.key(), value=transform(msg.value()))

        # the offsets become part of the transaction — this is the whole trick
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata(),
        )
        producer.commit_transaction()
    except Exception:
        producer.abort_transaction()
        raise

Three settings carry the guarantee:

• transactional.id — stable per logical instance. On restart, Kafka fences the previous session (zombie fencing), so a hung old instance can't write. If two live instances share one transactional.id, they fence each other in a loop.
• send_offsets_to_transaction — makes the consumer's offset commit part of the producer transaction. Without it you have atomic output but a separate offset commit, which is just at-least-once again.
• isolation.level=read_committed on downstream consumers — otherwise they read records from aborted transactions.

What transactions do not cover

This is the part people miss. The atomicity is Kafka-to-Kafka only:

producer.begin_transaction()
charge_credit_card(order)              # ← NOT part of the transaction
producer.produce("payments", ...)
producer.commit_transaction()          # if this aborts, the card is still charged

Any external side effect — a database write, an HTTP call, an email — is outside the transaction boundary. If you need atomicity across Kafka and a database, you need the transactional-outbox pattern (track 10) or an idempotent handler, not Kafka transactions.

The cost

Transactions add latency (each commit is a two-phase round trip through the transaction coordinator), reduce throughput meaningfully, require read_committed consumers to wait for the LSO (last stable offset) — so a long-running transaction blocks downstream reads — and add operational surface via transactional.id lifecycle management.

Decision rule: use transactions when the work is genuinely Kafka-to-Kafka (stream processing, enrichment, routing) and duplicates are unacceptable. Use at-least-once with idempotent handlers for everything that touches the outside world. Most services are the second case.

Command reference

Hands-on exercises

Use the broker from module 00.

kt --create --topic tx-in --partitions 2
kt --create --topic tx-out --partitions 2

1. Produce at-most-once loss

Commit before processing, then kill -9 the consumer mid-batch. Restart it.

Expected: records that were in flight are never processed and never redelivered — the offset moved past them. Count what you processed vs. what was produced and confirm the gap.

2. Produce at-least-once duplicates

Move the commit after processing and repeat the kill.

Expected: on restart, records already processed are delivered again. Log processed IDs to a set and count duplicates.

3. Show auto-commit is neither

Set enable.auto.commit=True, auto.commit.interval.ms=5000, and make processing take 8 seconds per batch. Kill mid-batch.

Expected: some records both committed and unprocessed (loss), depending on where the timer fell. Then make processing take 1 second and repeat — now you get duplicates instead. Same config, opposite semantics, decided by timing.

4. Prove the +1 matters

Commit msg.offset() instead of msg.offset() + 1, restart the consumer, and observe.

Expected: exactly one record per partition is reprocessed every restart. Small, permanent, easy to miss in testing.

5. Build an idempotent handler

Add a SQLite (or Postgres) processed_events(event_id PRIMARY KEY) table and the ON CONFLICT DO NOTHING guard. Re-run exercise 2's kill test.

Expected: duplicates are still delivered, but the business effect happens once. Confirm by counting rows in your effect table, not messages consumed.

6. Run a real transactional consume-process-produce

Implement the transaction loop above. Verify it works, then introduce a deliberate failure after produce() but before commit_transaction() (raise an exception so abort_transaction() runs).

Expected: with a read_committed consumer on tx-out, the aborted records are not visible, and the offsets weren't committed either — so on restart the input is reprocessed cleanly. Now read tx-out with isolation.level=read_uncommitted and confirm you can see the aborted records. That contrast is the whole point of isolation.level.

7. Watch zombie fencing

Start two processes with the same transactional.id. Have both attempt transactions.

Expected: one gets fenced with a ProducerFenced-style error the moment the other calls init_transactions(). Write one sentence on why this is a safety feature rather than a bug, and what it implies for how you assign transactional.id in Kubernetes.

8. Diagnose and fix: the double-charged customers

A payments consumer uses transactions and looks correct:

producer = Producer({"bootstrap.servers": "...", "transactional.id": "payments"})
producer.init_transactions()
consumer = Consumer({"group.id": "payments", "enable.auto.commit": True})

while True:
    msgs = consumer.consume(100, timeout=1.0)
    producer.begin_transaction()
    for msg in msgs:
        charge_card(msg)                                  # external API call
        producer.produce("receipts", value=receipt(msg))
    producer.commit_transaction()

Customers report duplicate charges after every deploy. Find all the bugs.

9. Clean up

kt --delete --topic tx-in; kt --delete --topic tx-out

Independent challenge

No solution given. You're building an order-fulfilment consumer that, per order event, must: write a row to Postgres, call a warehouse REST API to reserve stock, and emit a stock.reserved event to Kafka. Duplicates cause double-reservations, which cost real money.

Design the delivery guarantee. Decide whether Kafka transactions help at all here, and if not, what does. Specify exactly where idempotency keys live, what happens when the warehouse API times out (you don't know if it succeeded), and what your consumer does on restart mid-order. Then state the one failure mode your design still can't prevent, and why accepting it is reasonable.

Common mistakes & troubleshooting

• Assuming auto-commit gives at-least-once. It's timer-based and gives whichever semantic the timing happens to produce.
• Committing msg.offset() instead of offset + 1. One reprocessed record per partition on every restart.
• Believing transactions cover external side effects. They are Kafka-to-Kafka only; database writes and API calls are outside the boundary.
• Transactions without send_offsets_to_transaction(). Atomic output, non-atomic offsets — i.e. at-least-once with extra latency.
• Leaving enable.auto.commit=True with transactions. Defeats the offset guarantee entirely.
• Downstream consumers on read_uncommitted. They see records from aborted transactions.
• Sharing one transactional.id across instances. They fence each other continuously.
• Long-running transactions. They hold the LSO back and stall read_committed consumers downstream.
• Reaching for EOS when an idempotent handler would do. Simpler, faster, and covers external effects that transactions can't.

Checkpoint quiz

Further reading & sources

• KIP-98: Exactly-once delivery and transactional messaging - the transaction coordinator design, fencing, and LSO.
• Confluent: Transactions in Apache Kafka - the canonical walkthrough of consume-process-produce.
• Kafka: Consumer configuration — isolation.level - read_committed semantics and the LSO.
• confluent-kafka-python: transactional API - init_transactions, send_offsets_to_transaction and error handling.
• Track 10: Idempotency in practice - the idempotency-key pattern this module recommends over transactions for external effects.

Next

03-schemas-and-evolution-with-schema-registry — your records now arrive exactly as intended. Module 03 covers keeping them readable as producers and consumers change independently over years.