Exactly-once: checkpoint plus transactional sink

Why is the checkpoint the commit point of a stream job, and what does that cost in latency and timeouts?

What a stream processor provides is exactly-once effect, not delivery: duplicates still happen and are made harmless (exactly-once is about effects). Here the job that never ends adds one thing: the checkpoint is the commit point.

The mechanism

It composes three things:

The checkpoint is a consistent cut containing source offsets, so state and input position recover together.
The sink joins a two-phase commit, driven by the checkpoint's life cycle:
- pre-commit when the barrier passes the sink (the sink flushes its pending output into an open transaction, but does not commit it);
- commit when the engine announces checkpoint complete (all operators have saved their state);
- abort, and recover pending transactions, on restore.
After a crash, the job restores the last complete checkpoint. Output written after it was in an uncommitted transaction and is rolled back. Records are then replayed from the saved offsets. Each output appears exactly once in committed form.

          records ...|barrier n|... records ...|barrier n+1|...
Sink:     [ transaction n open: pre-commit at barrier n ]
          [ commit when checkpoint n completes ]  -> output becomes visible
          crash before completion -> abort, restore checkpoint n-1, replay

What the checkpoint as commit costs

The general costs of a transactional sink are fault-tolerance's. In a stream job two of them are set by the checkpoint.

1. Output is invisible until the checkpoint completes. So end-to-end latency is at least the checkpoint interval: with a 10-second interval, results appear 10 to 20 seconds late. (Ana's 30 seconds is her checkpoint interval plus duration.)

2. The transaction must outlive the checkpoint and the recovery. The sink holds an open transaction for the whole interval, so its timeout must exceed the maximum checkpoint duration plus the worst-case recovery time.

Two timeouts configured independently, with a hidden ordering constraint between them, is a classic production failure.

Where the sink can upsert by a key derived from the input, idempotent writes need no transaction at all and remove both costs (the three sources of idempotence). Either way the replayed computation must be deterministic, or the output differs on recovery and nothing detects it (determinism).

Case, resolvedConstructed case · Ana at Ridelo

Checkpoints from 20 seconds to 9 minutes

With 500 GB of state, 20-second checkpoints grew to 9 minutes under load, a 10-minute outage took 90 minutes to catch up, and the billing sink lost a transaction to a timeout. Where it started: Checkpoints and exactly-once →

Slow checkpoints. She moves to unaligned checkpoints for the busiest job, accepting larger checkpoints and more write bandwidth. She also moves to incremental checkpoints with the LSM backend, so each checkpoint uploads only changed files, and she watches the retention and consolidation of checkpoint files.
The 90-minute catch-up. She computes the catch-up equation and sees that running at 90% of capacity means any 10-minute outage costs 90 minutes. She sets autoscaling so the job normally runs at 50–60%, where the same outage costs about 10 minutes. She tells management that the unused 40% is the recovery budget.
The billing sink. She sets the transaction timeout to 30 minutes, above checkpoint duration plus worst-case recovery, and documents the ordering constraint. For the charges table, she then switches to idempotent upserts keyed by charge_id derived from the order ID. This removes the 2PC delay and the timeout coupling. She audits the code for random() and now() calls.