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.

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, but not in the result.

Ubik gets there with three things.

·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.

The checkpoint is one atomic write: the source offsets and the operator state land together, or not at all. kill -9 at any instant; resume reads the unit, rewinds Kafka to the recorded offsets, replays, and the sink carries the same bytes, nothing lost, nothing doubled. A stream-stream join rides the same unit: both source offsets, both watermark clocks, the two stores and the emit cursor. Proven byte-identical to the DuckDB batch oracle across kill -9 mid-window and mid-band.

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, and it does the same to a stream-stream join killed mid-band.

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

A checkpoint is one atomic write: every source offset and every operator's state, or nothing. A kill -9 takes only what accumulated after the last one. Resume reads the unit, rewinds the log to the offsets it recorded, replays the tail and carries on, so the run writes the bytes an uninterrupted run would have written.

Where the guarantee ends

·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 counted and reported, then dropped, never dropped in silence.
·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. On a metering pipeline that means a deduplication layer, a database with state and operations of its own.