At-Most-Once, At-Least-Once, and Exactly-Once

“Delivered” is not one thing. Kafka lets you choose which failures you would rather have, and the choice is made through configuration on both the producer and the consumer.

At-most-once: lose records, never duplicate them

The consumer commits its offset before doing the work. If it crashes mid-processing, the offset has already moved on and the record is never retried.

import time
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "at-most-once",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders"])

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    consumer.commit(msg)                      # commit FIRST
    print(f"committed @{msg.offset()}, now processing")
    time.sleep(2)
    if msg.offset() == 2:
        raise RuntimeError("crash mid-processing")
    print(f"  processed @{msg.offset()}")

$ python at_most_once.py
committed @0, now processing
  processed @0
committed @1, now processing
  processed @1
committed @2, now processing
Traceback (most recent call last):
  ...
RuntimeError: crash mid-processing

$ python at_most_once.py
committed @3, now processing
  processed @3

Offset 2 was committed but never processed, and it is gone for this group. Choose this only when stale data is worse than missing data — a metrics stream where you would rather drop a sample than double-count it.

At-least-once: never lose, sometimes duplicate

Commit after the work. This is the default posture for almost every pipeline.

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "at-least-once",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders"])

seen = []
while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    print(f"processing @{msg.offset()}")
    seen.append(msg.offset())
    time.sleep(2)
    if msg.offset() == 2 and seen.count(2) == 1:
        raise RuntimeError("crash after processing, before commit")
    consumer.commit(msg)                      # commit LAST
    print(f"  committed @{msg.offset()}")

$ python at_least_once.py
processing @0
  committed @0
processing @1
  committed @1
processing @2
Traceback (most recent call last):
  ...
RuntimeError: crash after processing, before commit

$ python at_least_once.py
processing @2
  committed @2
processing @3
  committed @3

Offset 2 was processed twice. Nothing was lost, and that is the trade. Your processing has to tolerate it — write with INSERT … ON CONFLICT DO UPDATE, or key your writes by event ID so a repeat is a no-op.

Exactly-once: transactions

For a consume-process-produce loop that stays inside Kafka, a transaction makes the output writes and the offset commit atomic.

from confluent_kafka import Consumer, Producer, KafkaError

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "eos-processor",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,          # required — the transaction commits offsets
})
consumer.subscribe(["orders"])

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "order-enricher-1",   # must be stable across restarts
    "enable.idempotence": True,
})
producer.init_transactions()

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    producer.begin_transaction()
    try:
        enriched = msg.value().decode().upper()
        producer.produce("orders-enriched", key=msg.key(), value=enriched)

        # The offset commit joins the transaction rather than going through the consumer.
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata(),
        )
        producer.commit_transaction()
        print(f"committed txn for @{msg.offset()} -> {enriched}")
    except Exception as e:
        producer.abort_transaction()
        print(f"aborted txn for @{msg.offset()}: {e}")

$ python eos.py
committed txn for @0 -> ESPRESSO
committed txn for @1 -> CORTADO
committed txn for @2 -> FLAT WHITE

If the process dies between produce and commit_transaction, the transaction is aborted by the broker and neither the output record nor the offset advance survive. On restart the input record is reprocessed and written exactly once.

The reader has to opt in

An aborted transaction still writes bytes to the log — the broker marks them aborted rather than removing them. A consumer only hides those records if you ask it to:

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "downstream",
    "auto.offset.reset": "earliest",
    "isolation.level": "read_committed",     # default is read_uncommitted
})

Compare the two on a topic where one transaction was aborted:

bin/kafka-console-consumer.sh --topic orders-enriched --from-beginning \
  --consumer-property isolation.level=read_uncommitted \
  --max-messages 10 --bootstrap-server localhost:9092

ESPRESSO
CORTADO
ROLLED-BACK-RECORD
FLAT WHITE
Processed a total of 4 messages

bin/kafka-console-consumer.sh --topic orders-enriched --from-beginning \
  --consumer-property isolation.level=read_committed \
  --max-messages 10 --bootstrap-server localhost:9092

ESPRESSO
CORTADO
FLAT WHITE
Processed a total of 3 messages

Exactly-once on the write side is undone by a reader left on the default read_uncommitted. Both halves have to be configured.

Choosing

The honest default is at-least-once with idempotent downstream writes. Reach for transactions when the processing genuinely cannot be made idempotent and both ends of the pipeline are Kafka.

Practice

ok  partition=0 offset=-1
ok  partition=0 offset=-1

FATAL: Local: This instance has been fenced by a newer instance

Next: giving your records a schema, so producers and consumers can evolve independently.