In Flink, how do checkpoint barriers produce a consistent snapshot across parallel operators?

## The problem barriers solve A Flink job is a graph of parallel subtasks spread over many machines, each processing records at its own speed. To recover from failure you need a snapshot in which every subtask's state corresponds to the *same* prefix of the input. The naive approach — pause every task, snapshot, resume — would stall the pipeline for as long as the slowest state write takes. Barriers get the same consistency without a global pause. ## What a barrier is A checkpoint barrier is a lightweight control record injected into the data stream and carrying a checkpoint ID. It flows through exactly the same channels as ordinary records and, crucially, is never reordered around them. That single property is what makes it a *cut*: for any channel, everything before the barrier belongs to checkpoint N, everything after it does not. The sequence is: 1. The **checkpoint coordinator** on the JobManager decides it is time for checkpoint N and instructs every source subtask to trigger it. 2. Each source subtask snapshots its own state — its current read positions — and emits a barrier N into its output. 3. Barriers propagate downstream with the records. 4. Every task that has seen barrier N on all of its inputs snapshots its state, forwards barrier N, and acknowledges N to the coordinator with a handle to the persisted state. 5. When all subtasks have acknowledged, the coordinator writes the checkpoint metadata, marks N complete, and calls back into the tasks to tell them so. ## Alignment A task with more than one input channel — the parallel instances feeding a keyed operator, or the two sides of a join — will receive barrier N on its channels at different times, because upstream tasks run at different speeds. In `EXACTLY_ONCE` mode — the default in Flink 2.3 — the task **aligns**: the channel that already delivered barrier N is blocked and its subsequent records are buffered; channels that have not yet delivered it keep being consumed. When the last barrier arrives, the task's state reflects precisely the pre-barrier records from every channel — no more, no less — so it snapshots, forwards the barrier, and unblocks the buffered channels. Alignment is where the cost lives. The task idles on the fast channels while it waits for the slow one, and the buffered records occupy network memory. Under backpressure, when a barrier crawls through queues of already-buffered data, alignment can dominate checkpoint duration. ## The at-least-once alternative Set `CheckpointingMode.AT_LEAST_ONCE` and tasks stop blocking channels. A multi-input task still waits until barrier N has arrived on every input before it snapshots and forwards the barrier, but in the meantime it keeps consuming the channels that already delivered it, so records from *after* the barrier on those channels are processed into state. The snapshot is therefore *ahead* of the cut on those channels. Recovery still rewinds the sources to the checkpoint's positions, so those records are processed a second time — hence at-least-once. In exchange, the fast channels never stall. A task with a single input has nothing to align, so a job made only of one-to-one operators (`map`, `filter`, `flatMap`) behaves exactly-once even in this mode. This is a correctness-versus-latency choice, not a bug. ## Asynchronous state writing Snapshotting does not mean the task blocks while gigabytes are uploaded. The task takes a fast, in-memory copy-on-write or immutable view of its state, forwards the barrier, and continues processing while the write to checkpoint storage happens on a background thread. Only when that write finishes does the acknowledgement go out. This is why the reported *duration* of a checkpoint splits into synchronous, asynchronous and alignment components in the web UI, and why reading those three numbers separately is the standard first diagnostic step. ## Completion and the callback A checkpoint is complete only when *every* subtask has acknowledged it — one slow or stuck subtask means no completed checkpoint at all, and the recovery point silently ages. On completion the coordinator invokes the completion callback on the tasks. That callback is not decoration: it is exactly where a two-phase-commit sink commits its staged transaction, because completion is the first moment the job knows the snapshot it belongs to will never be rolled back. ## Relationship to Chandy-Lamport The algorithm is the classic Chandy-Lamport distributed snapshot adapted for acyclic dataflow with a known topology, published as *asynchronous barrier snapshotting*. The differences from the textbook version matter in interviews: Flink's sources are the only injection points, alignment replaces recording in-flight channel contents, and the coordinator holds the completion decision centrally rather than deriving it from marker exchange. Enabling unaligned checkpoints moves Flink *back* toward the textbook algorithm by persisting in-flight buffers instead of draining them.