Where the Checkpoint Goes: Offset Ordering and Two Failure Directions

A streaming consumer does two things that must both succeed or the pipeline is wrong: it reads a record, and it writes a derived result. Between those two events there is a window, and the consumer’s stored position can be on either side of it. There is no third option. Every guarantee a pipeline advertises is a statement about which side of that window the position sits on, and what the pipeline does when a crash lands inside it.

The choice is usually presented as at-least-once versus at-most-once, with exactly-once as a third. That framing hides the interesting part. The two primitive orderings are simple; the difficulty is that a pipeline has several positions, namely the read position, the sink’s transaction, and the downstream consumer’s position, and they must be ordered relative to each other, not just to the write.

Two orderings, two losses

Take a consumer that reads record $r$ at offset $n$, transforms it, and inserts a row into a sink.

Commit then write. The consumer advances its position to $n+1$ and then performs the insert. A crash between the two leaves the position past a record whose effect never happened. The record is skipped on restart, permanently. This is at-most-once: no duplicates, possible gaps.

Write then commit. The insert completes, then the position advances. A crash between the two leaves the position at $n$ while the insert is durable. On restart the record is read again and inserted again. This is at-least-once: no gaps, possible duplicates.

The asymmetry is total. The first ordering loses data silently and cannot be repaired without going back to the source; the second duplicates data visibly and can be repaired by making the write idempotent. Since a sink write can be made idempotent with a key and a uniqueness constraint while a skipped record cannot be recovered without a rescan, at-least-once is the default choice for essentially every pipeline.

But “make the write idempotent” is doing a lot of work in that sentence, and its cost is the subject of everything that follows.

The commit is itself a write

The position has to be stored somewhere durable, and storing it is a write with the same failure properties as the sink write. If the position lives in a different system from the sink, such as an offset store, a coordination service, or a metadata table, then there are two writes and a crash can land between them. The ordering argument recurses: whichever write happens second determines the failure direction.

If the position write follows the sink write, a crash between them duplicates on restart, which is the safe direction. If it precedes, the pipeline can lose. So the position must be written after the sink commit, and the position store’s own durability must be at least as strong as the sink’s.

The problem is that two separate durable writes cannot be atomic without a distributed transaction, and distributed transactions across a stream processor and a sink are expensive and operationally fragile. The standard escape is to put the position inside the sink’s transaction.

Folding the position into the sink write

If the sink is a database, the consumer can write the derived rows and the new position in the same transaction:

BEGIN;
INSERT INTO orders_flat (order_id, customer_id, total_cents, source_offset)
VALUES (4711, 92, 129900, 88213)
ON CONFLICT (order_id) DO UPDATE
  SET total_cents = EXCLUDED.total_cents,
      source_offset = EXCLUDED.source_offset
  WHERE orders_flat.source_offset < EXCLUDED.source_offset;

UPDATE ingest_position SET last_offset = 88213 WHERE stream = 'orders';
COMMIT;

Now the position advance and the data write either both happen or neither does. A crash before the commit leaves the position at the previous value and the row absent; on restart the record is read and applied again, with no duplicate because the insert is keyed. A crash after the commit leaves both present, and the restart resumes from the new position.

Two details make this work. The ON CONFLICT clause must be conditioned on the incoming offset being newer than the stored one, otherwise replaying an older record after a newer one would overwrite newer data with older. And the position row must be updated with the same value the sink rows carry, so that a partially applied batch is detectable: after a crash, the position says 88213 and any row stamped above 88213 is evidence of a bug rather than a legitimate state.

The cost is that the sink must support multi-row transactions and the consumer must be able to batch a whole micro-batch into one transaction. For sinks that accept single-row writes only, such as a REST API or an append-only object store, this option is unavailable, and the two-write problem returns.

Batching multiplies the window

A consumer that commits per record pays a durable write per record for the position. A consumer that commits per batch of 500 pays one. The batch is the unit of atomicity, so the window between the read of the batch and the commit of the batch is where a crash can duplicate 500 records rather than one.

Batch size therefore sets the maximum duplicate volume after a crash, and it interacts with the sink’s own limits: a batch that exceeds the sink’s maximum transaction size is split, and each split commits separately, which reintroduces partial batches. The position must be advanced per split, not per batch, or a crash between splits re-applies the whole batch and the split’s earlier part duplicates.

The same arithmetic applies to backfill, where the unit of work is a key range rather than a batch, and to the snapshot phase, where it is a chunk.

Why automatic committing is a trap

Many consumers offer a mode in which the position is advanced periodically without the application’s involvement. The interval is measured in time or in records, and the commit happens regardless of whether the application finished processing.

This is the commit-then-write ordering with extra steps. The position can advance past records whose writes are still in flight or failed, and the failure is invisible: the consumer’s own loop saw the records, the commit happened, and the error surfaced in the sink after the position had moved. A consumer in this mode is at-most-once, whatever the documentation says about delivery.

The same mode also makes the commit interval a tuning knob with a perverse meaning: shortening it reduces duplicate volume after a crash and increases the write load on the position store; lengthening it does the reverse. Neither setting changes the direction of the failure, which is determined by the ordering alone.

Exactly-once as a property of a boundary

A pipeline can be made to deliver each record’s effect exactly once only if the read position and the sink write share a transaction boundary, as in the SQL example above. That is a real and common arrangement when the source and sink are the same engine or when the source is a log whose consumer can be given transactional write semantics.

The claim becomes fragile when the pipeline has more than one hop. A consumer that reads from a log, writes to a database transactionally, and then emits a change to another log has two boundaries, and a crash between them duplicates the emission even though the database write was exact. Exactly-once is a property of a single read-to-write boundary, and a chain of boundaries composes into at-least-once at the ends.

The honest formulation is that the pipeline is exactly-once with respect to the sink’s observable state, which requires the sink write to be keyed and the position to be in the same transaction. Anything a downstream consumer does with the emitted data reopens the question at its own boundary. Deletes and tombstones are where this matters most, because a duplicated delete is harmless while a duplicated insert is not, and the sink’s key semantics decide which one occurs.

Reading the position back

A consumer that restarts must trust the stored position. If the position store is the sink itself, the restart reads the position in the same transaction that it will use for the next batch, so there is no separate read to race with.

If the position is stored elsewhere, the restart reads it and then begins reading the source from there. A source that has discarded the records after that position, because retention expired as described in log-based capture and retention — cannot serve the restart. The consumer must detect this before processing: asking for an offset that no longer exists is an error, not an empty result, and treating it as an empty result silently loses everything in the gap.

The position is not the progress

A final distinction that trips up monitoring: the stored position measures what has been read, not what has been applied. In the write-then-commit ordering, the gap between the source’s end and the stored position is a lag that includes both unread records and records read but not yet committed. A pipeline whose commits are batched every 30 seconds reports a lag that oscillates by 30 seconds’ worth of records even when it is perfectly healthy.

Monitoring the sink’s own watermark, meaning the maximum source position present in the sink, measures applied progress and is the quantity that reconciliation can compare against the source. The stored position measures intent, and the two differ by exactly the in-flight batch.