Checkpoint Barriers and Aligned Snapshots

A checkpoint must capture the state of every operator at the same logical point in the stream. Taking each operator’s state independently does not work: an operator snapshot taken after it processed record n and its downstream snapshot taken before that record arrived describe a global state in which the record exists in one place and not the other. On recovery, the record is either lost or counted twice. The barrier protocol solves this by injecting a marker into the data stream itself, so that the marker’s position defines the cut and every operator sees the same one.

The cut, expressed in the data

A barrier is a special record that the source injects into each of its output channels at the moment a checkpoint is triggered. It flows through the graph in band, behind the records that preceded it and ahead of the records that follow. When an operator receives a barrier on an input, it knows that every record before the barrier on that input has been processed, and every record after it has not.

That is the local half of the cut. The global half is the observation that if every operator snapshots its state after processing all records before the barrier on all of its inputs, the resulting set of snapshots forms a consistent global state — one that corresponds to a point in the stream where no record is in flight between operators. This is the Chandy-Lamport construction applied to a dataflow graph, with the barrier playing the role of the marker message.

The protocol’s elegance is that it needs no global clock and no coordination during the snapshot. Each operator acts on the barriers it receives, and the consistency emerges from the property that barriers are never reordered relative to the data they accompany.

Alignment: waiting for all inputs

An operator with several inputs receives barriers from each of them at different times, because the channels carry different amounts of backlog. The operator cannot snapshot when the first barrier arrives, since the other inputs may still deliver records that precede the checkpoint. It must wait.

Waiting is called alignment. The operator stops consuming from the channels that have delivered their barrier and continues consuming from the channels that have not, buffering the arriving records. When the last barrier arrives, the operator processes the buffered records in order, snapshots its state, and forwards the barrier downstream on all outputs.

input A: --a1--a2--[B]--a3--
input B: --b1--b2--b3--b4--[B]--b5--

at the operator: consume b3, b4 (they precede the barrier);
snapshot after b4; forward [B] downstream.

The correctness requirement is that the operator forwards exactly one barrier per checkpoint on each output, and that it does so only after snapshotting. Forwarding before snapshotting lets a downstream operator snapshot state that includes records the upstream has not yet recorded, which breaks the cut.

The cost of alignment is the stall. The channel that delivered its barrier first is paused for as long as the slowest channel takes, and during that time its records accumulate in the buffer. A job with one chronically slow input — a join against a backlogged side, a union with a lagging partition — stalls every checkpoint for the duration of the slowest channel’s backlog. The buffered records are also state, briefly, and their volume is the backlog difference between the fastest and slowest channel.

Unaligned checkpoints and the stall they remove

The stall is severe under backpressure, which is exactly when checkpoints are most likely to be needed. A job whose channels are full has a large backlog difference, so alignment pauses the fast channels for a long time, which increases the backlog further, which lengthens the next alignment. Under sustained backpressure, aligned checkpointing can fail to complete at all.

Unaligned checkpointing removes the wait by snapshotting the buffers as well. When a barrier arrives on one input, the operator snapshots its state immediately and records the contents of the other channels’ buffers as part of the checkpoint, including the in-flight records. Recovery replays those buffered records after restoring the state, which restores the cut without having paused the stream.

The cost moves from time to space: the checkpoint now contains the in-flight data, so its size depends on how much is in flight, which under backpressure is large. A checkpoint that was a few megabytes of state can become hundreds of megabytes of buffer. The trade is deliberate — unaligned checkpointing completes quickly and recovers slowly, aligned checkpointing completes slowly and recovers quickly.

What the barrier must carry

A barrier carries the checkpoint identifier, which lets an operator distinguish barriers belonging to different checkpoint attempts and discard the state of an abandoned one. It also carries the checkpoint’s position in the source, either directly or by reference to the coordinator, so that recovery knows where to resume reading.

The position is the part that interacts with sinks. A checkpoint’s completion is what allows a transactional sink to commit, as described in Two-Phase Commit and Exactly-Once Sinks, so the barrier’s arrival at every sink is the trigger for the commit phase. A barrier that never reaches a sink because the job failed means the corresponding transaction is aborted, which is the correct behaviour: the checkpoint it belonged to was never completed.

Watermarks also travel with checkpoints. The watermark position at the cut is part of the snapshot, because resuming with a reset watermark would re-close windows that already emitted. Watermarks as Completeness Signals treats this as a requirement of the watermark’s semantics; here it is one more item in the snapshot’s contents, alongside operator state and timers.

Incremental and full snapshots

A full snapshot serializes every byte of every operator’s state at each checkpoint. Its cost is proportional to total state size, so it scales badly for jobs with large retained state and short checkpoint intervals.

An incremental snapshot records only what changed since the previous checkpoint, typically by capturing the set of state files created or modified. For a disk-backed store this is natural: the store’s files are immutable once written, so a new checkpoint references the previous files plus the new ones, and the snapshot is a manifest of file references rather than a copy of data. For an in-memory backend the same idea requires tracking which keys were modified, which is more intrusive and less commonly implemented.

The cost of incremental snapshots is on recovery: the job must reconstruct the full state from a chain of increments, so recovery time grows with the number of increments retained. The retention policy that bounds the chain is a trade between storage and recovery time, and a chain of a thousand increments can take longer to restore than the failure it recovers from.

# Reconstructing state from a chain of incremental manifests.
def restore(manifests):
    live = {}
    for m in manifests:                 # oldest first
        for path, meta in m["files"].items():
            if meta["deleted"]:
                live.pop(path, None)    # tombstone from a later increment
            else:
                live[path] = meta       # newer version wins
    return live

The tombstone handling is where incremental restore goes wrong. A file deleted in increment n and absent from n+1 is still deleted, and a naive union of all manifests resurrects it. The manifest must record deletions explicitly, and the restore must apply them in order.

Consistency across the graph

The barrier protocol guarantees a consistent cut only if every operator participates. An operator that snapshots asynchronously, or that forwards a barrier before its own snapshot completes, breaks the guarantee for the whole job. The failure is not local: a downstream operator’s snapshot can include a record whose upstream state was captured before it was processed, and recovery then produces a duplicate or a gap that is very hard to attribute to its cause.

Two operator types are especially easy to get wrong. A source must inject the barrier at a position it can resume from, which for a partitioned source means injecting into every partition at the corresponding offset; a source that injects at a global point but resumes from per-partition offsets restores an inconsistent read position. A sink must not commit before the checkpoint completes, which means its own state must record the pending transaction as part of the snapshot.

The resulting picture is a distributed snapshot in the classical sense, taken without stopping the system, whose cost is dominated by state size and whose correctness depends on every participant following the same protocol. The mechanism is well specified and the implementations are mature; the failures that remain come from operators outside the protocol — external systems that cannot participate, and custom operators that snapshot without honouring the barrier’s position.