What is At-least-once?

How it works

A consumer does two things for every batch: it processes the records (writes them somewhere) and it commits its offset, telling the broker “I am done up to here”. The order of those two steps decides the guarantee:

• Commit, then process: a crash loses the batch. That is at-most-once.
• Process, then commit: a crash replays the batch. That is at-least-once.

At-least-once is the safe default because lost data is usually worse than duplicate data, and duplicates can be removed. In Kafka it means turning off auto-commit (or keeping it but committing only after processing) and committing offsets after the write succeeds. Producers add the same risk on their side: a retry after a lost acknowledgment can write a message twice unless the idempotent producer is enabled.

To get correct results, the sink must absorb the replays. Common approaches are an upsert on a unique key, a primary key that rejects repeats, a dedup step keyed on an event ID, or a transaction that stores the offset with the output, which upgrades the pipeline to exactly-once results.

Example

The consumer crashes after writing pay-2 but before committing offset 2, so the restart writes pay-2 again. Summing naively double-counts; keying by payment ID gives the right total (Python 3.11).

events = [(0, "pay-1", 40), (1, "pay-2", 25), (2, "pay-3", 10)]  # (offset, id, amount)
committed = 0          # the consumer's committed offset
sink = []              # an append-only target

def run(crash_after_write=None):
    global committed
    for offset, pid, amount in events[committed:]:
        sink.append((pid, amount))              # 1. process / write
        if offset == crash_after_write:
            raise RuntimeError("crash before offset commit")
        committed = offset + 1                  # 2. then commit the offset

try:
    run(crash_after_write=1)
except RuntimeError as e:
    print(e)
run()                                           # restart from the committed offset: pay-2 again
print("rows written:", [p for p, _ in sink])
print("naive total:", sum(a for _, a in sink))
print("deduplicated total:", sum(dict(sink).values()))   # keyed by payment id

crash before offset commit
rows written: ['pay-1', 'pay-2', 'pay-2', 'pay-3']
naive total: 100
deduplicated total: 75

Common mistakes

• Leaving Kafka auto-commit on with asynchronous processing, which can commit offsets for records not yet written (at-most-once by accident).
• Appending to a table with no unique key, so every replay doubles some rows.
• Deduplicating on a key that is not stable across retries, such as a processing timestamp.
• Treating rare duplicates as harmless in money or count metrics that downstream teams trust.