Exactly-once in streams

A stream processor keeps state, such as per-key counts or the set of IDs it has seen. After a crash, that state must match the input position it was read from. Otherwise the results drift. Flink and Kafka Streams solve the state part with checkpoints. The output part needs one more step from the sink.

The big idea

Think of reading a book with a bookmark. You save your notes and the page number together. If you lose your place, you go back to the bookmark and read again. Pages after the bookmark are read twice, but the notes were saved only at the bookmark, so they are not doubled.

The bookmark protects the state. It does not protect things you already sent out, such as messages or database writes. Those must be safe to repeat.

Walking through it

A counter has checkpointed count = 1,200 at Kafka offset 5,000. It keeps processing and reaches offset 5,400, where count = 1,600. Then it crashes. Nothing after offset 5,000 was saved.

Restore checkpoint:      count = 1,200, rewind input to offset 5,000
Replay 400 events:       count = 1,200 + 400 = 1,600
State:                   counted once, as if no crash happened
Outputs for 5,000-5,400: emitted before the crash, then emitted again

Flink does this with barriers. The job manager injects a barrier into each source, along with the source offsets. Each operator snapshots its state when the barrier reaches it, then passes the barrier on. When every sink acknowledges, the checkpoint is complete.

The state is counted once. The output may still repeat, so the sink must handle that:

• Idempotent sink. Upsert rows keyed by (ad_id, minute). Writing the same row twice leaves one row.
• Transactional sink. Flink's two-phase commit writes to Kafka inside a transaction that commits with the checkpoint. Consumers with read_committed never see uncommitted output. Kafka Streams offers exactly_once_v2 for this.

So "exactly-once" in streaming means effectively once: at-least-once delivery, plus an effect that is deterministic and repeat-safe.

Common mistakes

• Writing count = count + n to the sink. Every recovery adds the replayed events again.
• Calling the pipeline exactly-once without naming the sink. The guarantee stops at the sink.
• Unbounded state. Deduplication state with no time-to-live grows until checkpoints slow down.

Check yourself

1. In the example, what is the count after recovery? 1,600. The checkpoint holds 1,200, and replaying 400 events adds 400.
2. Why is an upsert keyed by (ad_id, minute) safe to repeat? Writing the same key with the same value again leaves one row, so repeats have no extra effect.

Key takeaways

• Checkpoints save state with its input position, so state is counted once after a crash.
• Replayed input can repeat outputs, so the sink needs to handle repeats.
• Idempotent upserts and transactional sinks are the two options.
• Effectively-once is at-least-once plus repeat-safe effects.

Further learning

Next, see how this design fits into two overall architectures in Lambda vs Kappa.

For the general rules behind repeat-safe writes, see idempotency.