Checkpointing

Analogy categoryAnalogies

Why Streaming Jobs Need Checkpoints

Because a Flink job runs indefinitely and holds state distributed across many parallel operators, a failure — a crashed task manager, a network partition — must not mean starting over from nothing. Checkpointing is Flink's mechanism for periodically taking a consistent, distributed snapshot of the entire job's state and its position in each input stream, so that on failure the JobManager can restart the job and restore every operator's state to exactly the point captured by the last completed checkpoint.

🏏

Cricket analogy: A day-night Test match uses stumps breaks and tea intervals as fixed checkpoints where the exact score, overs bowled, and player states are formally recorded, so if play is interrupted by rain, it resumes from that recorded point rather than from ball one.

The Chandy-Lamport Barrier Algorithm

Flink implements checkpointing using a variant of the Chandy-Lamport distributed snapshot algorithm: the JobManager periodically injects special checkpoint barrier records into the source operators' streams, which flow downstream interleaved with regular data. When an operator receives a barrier on all its input channels, it snapshots its current state to the configured checkpoint storage and forwards the barrier downstream, guaranteeing that the resulting snapshot reflects a consistent cut — everything processed before the barrier is included, everything after is excluded — without ever pausing the entire pipeline.

🏏

Cricket analogy: An umpire's signal for a drinks break travels down the field to every player at slightly different moments, but once every fielder has acknowledged it, play formally pauses at a consistent point — like a checkpoint barrier propagating through the pipeline.

java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
CheckpointConfig config = env.getCheckpointConfig();
config.setMinPauseBetweenCheckpoints(30000);
config.setCheckpointTimeout(120000);
config.setMaxConcurrentCheckpoints(1);
config.setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

Aligned vs. Unaligned Checkpoints

In the classic aligned checkpoint, an operator with multiple input channels must wait for the barrier to arrive on every channel before snapshotting — buffering data from channels whose barrier arrived early — which can stall throughput badly under backpressure since a slow channel delays the whole barrier alignment. Unaligned checkpoints, introduced in Flink 1.11, solve this by letting the barrier overtake buffered in-flight records: instead of waiting for alignment, the operator immediately snapshots and includes the in-flight records themselves as part of the checkpoint state, trading a larger checkpoint size for dramatically reduced checkpointing latency under heavy backpressure.

🏏

Cricket analogy: Waiting for every fielder to jog back to position before restarting play (aligned) versus the umpire just recording exact positions instantly and resuming immediately (unaligned) is the same latency-versus-completeness tradeoff Flink makes.

Unaligned checkpoints increase checkpoint size because in-flight buffer contents are persisted alongside operator state, and they are not compatible with certain features like watermark alignment across sources in the same simple way — evaluate the tradeoff rather than enabling them by default.

Checkpoint Storage and Failure Recovery

Once every operator has acknowledged its snapshot to the JobManager, the checkpoint is marked complete and becomes the recovery point; incomplete checkpoints (where an operator failed before acknowledging) are discarded. On failure, Flink restarts the affected tasks — or the whole job, depending on the configured restart strategy — and each operator reloads its state from the last completed checkpoint's location in checkpoint storage, while sources rewind to the recorded stream offsets, replaying exactly the records that occurred after that snapshot.

🏏

Cricket analogy: A match is only officially recorded as complete once the scorers, umpires, and both captains have all signed off on the final scorecard; an unsigned scorecard from an abandoned match is discarded, like an unacknowledged checkpoint.

Checkpoint storage is typically a durable, distributed filesystem like S3, HDFS, or GCS configured via execution.checkpointing.storage; for very small state, JobManagerCheckpointStorage keeps snapshots in the JobManager's heap, which is fine for local testing but unsuitable for production.

Checkpointing periodically snapshots a job's complete distributed state so it can recover after a failure without reprocessing from scratch.
Flink uses a Chandy-Lamport-style barrier algorithm: barriers flow through the stream and trigger local state snapshots when received on all input channels.
Aligned checkpoints wait for barrier alignment across all channels, which can stall under backpressure.
Unaligned checkpoints let barriers overtake in-flight data, snapshotting immediately at the cost of larger checkpoint payloads.
A checkpoint is only marked complete once every operator has acknowledged its snapshot to the JobManager.
On recovery, operators reload state from the last completed checkpoint and sources rewind to the recorded offsets.
Checkpoint storage should be a durable distributed filesystem (S3, HDFS, GCS) in production, not JobManager heap.