Why architecture matters here
Consider a payments pipeline computing per-merchant running totals and emitting settlement records. Under at-least-once semantics, a single TaskManager crash replays every record since the last checkpoint: totals double-count, settlement records duplicate, and money is wrong. Under at-most-once, the same crash silently drops in-flight records and money is wrong the other way. Every fraud counter, billing aggregate, inventory level, and CDC-derived table has this shape: the business value is the correctness of the accumulated state, and “approximately once” is just “wrong” with extra steps.
Architecture matters because exactly-once is a property of the whole path, not of any component. A perfectly consistent Flink checkpoint protects internal state, but if the sink writes to Kafka without transactions, downstream consumers still see the replayed duplicates. A transactional sink protects output, but if the source cannot rewind to the snapshotted offset, recovery has a hole. The guarantee is a chain — replayable source, snapshotted state, committed output — and it is exactly as strong as its weakest link. Most “exactly-once bugs” in production are one link quietly configured out of the chain: a sink left at NONE delivery guarantee, a transaction timeout shorter than a bad checkpoint, a side effect (an HTTP call, a metric, an email) performed inside an operator where replay will fire it twice.
Understanding the machinery is also what lets you price it. Exactly-once costs checkpoint bandwidth, sink commit latency quantized to the checkpoint interval, and operational care. For some pipelines — dashboards over idempotent upserts — at-least-once into an upsert sink is the same observable result at lower cost. Knowing when the full protocol is warranted is itself an architectural decision.
The architecture: every piece explained
Replayable source. Kafka consumers in Flink do not commit offsets to Kafka as their source of truth; the offsets live inside the checkpoint. On recovery, the connector seeks each partition to the snapshotted offset and replays. This is the foundation: reprocessing is always possible, so the problem reduces to making reprocessing harmless.
Barrier-based snapshots. The checkpoint coordinator (in the JobManager) periodically injects numbered barriers into every source partition. Barriers flow with the data. When an operator with multiple inputs receives barrier n on one input, it pauses (or buffers) that input until barrier n arrives on all inputs — alignment — guaranteeing its snapshot reflects exactly the records before the barrier on every channel. It then snapshots its state asynchronously (RocksDB SST files uploaded incrementally to S3/HDFS) and forwards the barrier. When every operator has acknowledged, the coordinator marks checkpoint n complete and notifies all operators. Unaligned checkpoints trade this cleanliness for speed under backpressure by snapshotting in-flight buffers too.
Two-phase-commit sink. The Kafka sink opens a transaction per checkpoint epoch. Records written during epoch n are produced into that transaction — durable on the broker but invisible to read_committed consumers. At barrier time the sink pre-commits: flushes everything and snapshots the transaction ID into the checkpoint. Only on the coordinator’s completion notification does it commit, atomically publishing the epoch via the broker’s transaction markers. Crash before completion means recovery aborts the dangling transaction; crash between completion and commit means recovery re-commits the snapshotted transaction ID (Kafka commits are idempotent per txn ID). Output visibility is thus tied to checkpoint success.
Idempotent sinks are the cheaper sibling: deterministic keys plus upsert semantics (Cassandra, Elasticsearch, JDBC upsert) make replayed writes overwrite themselves. No transactions, no commit latency — but temporarily visible “retracted” intermediate values during replay, which some consumers cannot tolerate.
End-to-end flow
Trace one checkpoint through a fraud-scoring pipeline: Kafka source, keyed enrichment, windowed aggregate, Kafka transactional sink, checkpoint interval 30s. (1) The coordinator triggers checkpoint 42. Each source task records its partition offsets (say p0:1,000,517, p1:998,204) into the snapshot and emits barrier 42 in-stream. (2) The enrichment operator receives barrier 42 on both of its input channels, aligns, snapshots its RocksDB state (only SSTs changed since checkpoint 41 — incremental), forwards the barrier, and resumes. (3) The window operator does the same for its timers and accumulators. (4) The sink receives barrier 42, flushes its open transaction txn-42 to the brokers, snapshots the transaction handle, and acknowledges — the pre-commit. Everything written this epoch sits on the brokers, invisible. (5) The coordinator, having heard from all operators, finalizes checkpoint 42 and notifies; the sink commits txn-42, the broker writes commit markers, and 30 seconds of output becomes visible to read_committed consumers atomically. The sink immediately opens txn-43.
Now the failure: at T+47s a TaskManager dies mid-epoch 43. (6) The JobManager cancels the job and restores every operator from checkpoint 42’s snapshots — RocksDB state re-downloaded, source offsets reset to p0:1,000,517/p1:998,204. The dangling txn-43 on the brokers is aborted (its records are skipped by committed readers forever). (7) The pipeline replays the 17 seconds of records processed since checkpoint 42. The window state absorbs them from the restored baseline — no double counting, because the state that had already absorbed them was discarded with the failed epoch. The sink writes them into a fresh transaction. (8) To a downstream read_committed consumer, the timeline is: epoch 42 appeared, a pause, epoch 43 appeared. No duplicate, no gap — the crash is observable only as latency. That is the whole guarantee: failures compress into pauses.