What is Exactly-once?

How it works

Networks fail between “work done” and “work acknowledged”, so a system can only choose how it fails:

• At-most-once: acknowledge first, then process. A crash loses records.
• At-least-once: process first, then acknowledge. A crash replays records, so duplicates appear.
• Exactly-once: at-least-once delivery plus a way to make the replay harmless.

There are two ways to make the replay harmless. The first is an idempotent sink: write with a unique key (an upsert or a primary key), so a replayed record overwrites itself instead of adding a row. The second is atomic commit of output and position: store the consumer offset in the same transaction as the result, so after a crash both roll back together and the restart resumes from exactly where the last commit ended.

Kafka implements the second pattern inside Kafka with idempotent producers and transactions (read_committed consumers see only committed output). Spark Structured Streaming does it with checkpointed offsets plus a replayable source and an idempotent or transactional sink, such as Delta Lake. The guarantee covers only the systems inside that transaction: a side effect outside it, such as sending an email, can still happen twice.

Example

A consumer stores its next offset in the same database transaction as each payment. It crashes while writing pay-2; the transaction rolls back, the restart resumes at offset 1, and every payment is counted once (Python 3.11, sqlite3).

import sqlite3

events = [(0, "pay-1", 40), (1, "pay-2", 25), (2, "pay-3", 10)]  # (offset, id, amount)
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE payments (id TEXT PRIMARY KEY, amount INT)")
db.execute("CREATE TABLE offsets (consumer TEXT PRIMARY KEY, next_offset INT)")
db.execute("INSERT INTO offsets VALUES ('billing', 0)")

def run(crash_after=None):
    (start,) = db.execute("SELECT next_offset FROM offsets WHERE consumer='billing'").fetchone()
    for offset, pid, amount in events[start:]:
        with db:  # one transaction: the row and the new offset commit together, or neither does
            db.execute("INSERT INTO payments VALUES (?, ?)", (pid, amount))
            if offset == crash_after:
                raise RuntimeError("crash before commit")
            db.execute("UPDATE offsets SET next_offset=? WHERE consumer='billing'", (offset + 1,))

try:
    run(crash_after=1)        # dies while writing pay-2
except RuntimeError as e:
    print("first run:", e)
run()                         # restart resumes from the stored offset
print(db.execute("SELECT id, amount FROM payments ORDER BY id").fetchall())
print("total:", db.execute("SELECT SUM(amount) FROM payments").fetchone()[0])

first run: crash before commit
[('pay-1', 40), ('pay-2', 25), ('pay-3', 10)]
total: 75

Common mistakes

• Assuming a Kafka setting alone gives end-to-end exactly-once when the sink is an external database outside the transaction.
• Committing consumer offsets before the write succeeds, which silently loses records on a crash.
• Appending with no unique key, so every replay adds duplicate rows.
• Calling non-idempotent side effects (emails, payments to a third-party API) inside a retried step.