The part that is hard

Exactly-once, and what that actually means

Exactly-once delivery does not exist. Over an unreliable channel you either send without retrying and lose messages, or retry until acknowledged and produce duplicates. There is no third option.

What is achievable is exactly-once effect: for a given input log and any sequence of failures, the output state is identical to what one failure-free run over the same log would have produced. Duplicates may exist on the wire. They may not change the result.

Ubik gets there with three things, and none of them is magic.

·The offset lives inside the checkpoint. One atomic file holds all operator state and all source offsets together, written to a temp file, fsynced, then renamed. There is no separate offset commit anywhere, so state and position cannot disagree.
·Replay is deterministic by construction. No wall-clock on the data path, no unordered iteration, no randomness. A group hashes to a fixed partition, updates keep input order, emission merges back key-ascending. A checkpoint taken at three workers resumes at sixteen unchanged.
·Re-writing the output is a no-op. Rows are keyed by (window, group) and upserted, so a window re-emitted after a crash writes byte-identical rows over the first.

Kill it mid-window

Resume from the checkpoint and the counters pick up where they were. Dump the same topic cold, run DuckDB batch over it, compare: the numbers match to the cent. That comparison is the test suite, run against a real broker on every merge.

shell

$ # SIGKILL mid-stream, then resume from the checkpoint$ ubik --from kafka://localhost:9092/txns \         --to   kafka://localhost:9092/txn_counts \         --resume ./txns.ckpt \      "SELECT merchant,              TUMBLE(event_time, INTERVAL 1 MINUTE) AS window,              count(*) AS n       FROM txns       GROUP BY merchant, TUMBLE(event_time, INTERVAL 1 MINUTE)"     # the resumed run is byte-identical to an uninterrupted one

Where the guarantee ends

Exactly-once holds between two named boundaries, so here are ours.

·It holds from the input log to the output topic. If your consumer reads a row and increments a counter somewhere without idempotence, the property stops at that hop.
·It covers events that arrive within their watermark. An event later than that is dropped, counted, and reported. Never dropped in silence, but dropped.
·It requires a replayable, immutable log upstream. A file source hashes the prefix it consumed and refuses to continue if the log changed underneath it.
·Windows close on the watermark and emit once. No retractions, no late corrections downstream.

The practical difference is who has to be idempotent. At-least-once pushes that onto every consumer, forever. On a metering pipeline that work has a name: a deduplication layer, which is a database, with state and operations. You have re-deployed a cluster to clean up after the engine that was meant to save you from one.