Distributed Snapshots: How Flink Checkpoints a Moving Stream

To back up a database you can, at worst, stop writes for a moment. A stream processor never gets that moment: events flow through dozens of operators across many machines, and "the state of the system" includes not just each operator's memory but the events in flight between them. Photographing that without stopping it sounds impossible — a camera panning across a scene that keeps moving. The Chandy-Lamport algorithm (1985) showed it isn't, and Flink's checkpointing is that idea industrialized. It is the prettiest algorithm that shows up in system design interviews, and most candidates have used it daily without knowing its name.

What "consistent" means here

A snapshot is consistent if it represents a state the system could have been in at a single logical instant: for every event, either its effects are fully included (in whichever operator's state absorbed it) or fully excluded (it will be replayed from the source). The failure mode to avoid: operator A's state reflects event e, operator B's doesn't, and e is also gone from the channel — e half-happened, and recovery double-counts or drops it. Exactly-once state semantics is precisely "recover to a consistent cut, then replay from it."

The marker trick

Chandy-Lamport's insight: flow a special marker through the same channels as the data, and let the marker be the boundary between "before the snapshot" and "after."

source: on checkpoint trigger -> record own offset -> emit barrier(n) into stream
operator with k inputs:
  barrier(n) arrives on input 1 -> pause input 1 (buffer its events)
  ... wait until barrier(n) arrives on ALL k inputs   <- alignment
  -> snapshot own state (async upload to S3)
  -> emit barrier(n) downstream, unpause inputs
sink: barrier(n) arrives -> ack; when ALL acks reach the coordinator,
      checkpoint n is complete (atomically: all states + source offsets)

Everything that entered the stream before the barrier is reflected in some snapshotted state; everything after will replay from the recorded source offsets. The in-flight-events problem dissolves: pre-barrier events are, by definition of the pause-and-align, absorbed into state before the operator photographs itself. (Classic Chandy-Lamport records channel contents instead of aligning; Flink's aligned variant trades a pause for empty-channel snapshots — smaller, simpler recovery.)

Recovery is the payoff for all this ceremony: restore every operator's state from checkpoint n, rewind sources to n's offsets, run. The replayed window is at-least-once internally, but since state reflects nothing past the barrier, effects apply exactly once in state. End-to-end exactly-once needs the sinks in on the trick — transactional sinks stage output per checkpoint and commit on checkpoint-complete (2PC keyed to barriers), or idempotent sinks absorb the replay. Same boundary theorem as always: exactly-once inside the snapshot domain, idempotency or transactions at the exits.

Where it hurts: alignment under backpressure

Alignment is a synchronization point, and its cost concentrates exactly when you can least afford it. A skewed join — one input fast, one slow — makes fast inputs buffer for seconds waiting for the laggard's barrier; under backpressure, barriers crawl through congested channels and checkpoints start timing out — and since longer checkpoints mean longer recovery replay, degradation compounds. The fixes are the modern half of the story:

• Unaligned checkpoints: don't wait — let barriers overtake buffered events, snapshotting the in-flight events themselves as part of state (full Chandy-Lamport, channel recording included). Checkpoint duration decouples from backpressure; the price is fatter checkpoints and channel state in recovery.
• Incremental checkpoints: with RocksDB state backends, upload only changed SSTs since the last checkpoint (immutable files again paying rent); a 1TB keyed state checkpoints in the delta, not the terabyte.
• Local recovery: keep a copy of the last snapshot on each worker's disk; restart-in-place skips the S3 download for the common single-node failure.

Checkpoint frequency is the operator's knob: interval ≈ recovery replay window; too frequent steals throughput (barriers + uploads), too rare makes every failure a long rewind. Production folklore: checkpoint duration creeping toward the interval is the earliest reliable smoke alarm for backpressure.

Savepoints — the same mechanism, triggered manually, with an operator-keyed portable format — are how running jobs get rescaled, upgraded, or forked: stop-with-savepoint, restart from it with different parallelism (keyed state reshuffles by key-group) or new code. The snapshot algorithm quietly became the deployment story, which is very Flink.

Interview probe	Answer sketch
Why not snapshot every operator at wall-clock T?	No common clock, and channels would hold un-attributed events — the cut wouldn't be consistent; markers define logical simultaneity without clocks
Two concurrent checkpoints?	Numbered barriers; an operator handles n while n+1's barriers queue — coordinator caps concurrency (usually 1-2)
State bigger than memory?	RocksDB backend: state is an LSM on local disk, checkpoints are SST uploads — the storage-engine post cashing in
Kafka transactions on the sink stall?	Uncommitted checkpoint = uncommitted output; consumers with read-committed see latency = checkpoint interval — say that coupling out loud

The durable idea: a globally consistent moment doesn't require stopping the world — it requires an agreed boundary flowing through the same pipes as the work. Barriers are that boundary; everything else (alignment, unaligned mode, incremental uploads) is engineering the cost of agreeing.