Definition
- Chandy-Lamport algorithm – Flink checkpointing is based on this distributed snapshot algorithm that captures a consistent cut of the dataflow without stopping processing.
- Checkpoint barriers – Special markers injected at sources that flow downstream through the graph alongside normal records, carrying a checkpoint ID.
- Barrier alignment – Multi-input operators wait until the barrier arrives on all input channels before snapshotting, ensuring the snapshot is consistent.
- Asynchronous snapshots – State is captured and uploaded to durable storage (S3/HDFS) in the background, so the processing thread is barely blocked.
- Coordinator acknowledgment – When all operators acknowledge their snapshots, the checkpoint is marked complete and usable for recovery.
The National Bank of India wants a perfectly consistent photograph of its entire assembly line of clerks at one logical instant, without stopping the line. The checkpoint coordinator sends a special colored slip (barrier) down each conveyor. Each clerk keeps processing normal slips until the colored slip arrives. A clerk fed by two conveyors waits until the colored slip has arrived on both before taking a photo of their passbook state. Once every clerk has filed their photo, the bank has one coherent snapshot.
Understanding the Concepts
How does Flink trigger a checkpoint?
The checkpoint coordinator lives in the JobManager and triggers a checkpoint on a configured interval. To start, it tells every source to inject a barrier with a new checkpoint ID into its output stream.
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// Enable checkpointing every 10 seconds
env.enableCheckpointing(10_000);
// Configure checkpoint storage
env.getCheckpointConfig().setCheckpointStorage(
"s3://nbi-flink-checkpoints/");
// Set exactly-once mode
env.getCheckpointConfig().setCheckpointingMode(
org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);
Checkpoint trigger flow:
JobManager (Coordinator)
│
├──▶ Source 1: inject barrier(id=42)
├──▶ Source 2: inject barrier(id=42)
└──▶ Source 3: inject barrier(id=42)
What are checkpoint barriers?
A barrier is not data; it is a marker that separates records belonging before the checkpoint from those belonging after it. Barriers flow downstream exactly like records, never overtaking them, so the ordering is preserved.
// Conceptual stream content (barriers are injected by the source):
// [record] [record] [record] [BARRIER-42] [record] [record] ...
// ◀── before CP-42 ──▶ ◀── after CP-42 ──▶
Barrier flow through the graph:
Source ──▶ [R1] [R2] [B42] [R3] ──▶ Map ──▶ [R1'] [R2'] [B42] [R3']
──▶ Sink
Barrier B42 separates pre-CP and post-CP records
Barrier never overtakes records: ordering preserved
How does barrier alignment work?
When a multi-input operator (after a join or union) receives the barrier on one input, it stops consuming that channel and buffers further records from it, while continuing the other channels until their barriers arrive. Only then does it snapshot.
// Multi-input operator: join of retail and corporate streams
// Input 1 (retail): [R1] [R2] [B42] [R3] [R4]
// Input 2 (corporate): [C1] [C2] [C3] [C4] [B42]
// Step 1: B42 arrives on Input 1 first
// - Stop consuming Input 1, buffer R3, R4
// - Continue consuming Input 2: process C1, C2, C3, C4
// Step 2: B42 arrives on Input 2
// - All barriers received
// - Snapshot state (exactly all pre-B42 records processed)
// - Forward B42 downstream
// - Drain buffered R3, R4 from Input 1
How does asynchronous snapshotting work?
When an operator snapshots, it captures a copy of its state and hands it to a background thread that uploads it to durable storage. The main processing thread is barely blocked.
// Snapshot is asynchronous:
// 1. Operator takes a lightweight copy-on-write snapshot
// 2. Background thread serializes and uploads to S3
// 3. Processing thread continues with new records
// 4. When upload completes, operator ACKs to coordinator
// Checkpoint storage configuration:
env.getCheckpointConfig().setCheckpointStorage(
"s3://nbi-flink-checkpoints/job-fraud-scoring/");
// Snapshots uploaded here asynchronously
How does recovery work?
On failure, Flink restarts the job, restores every operator's state from the last completed checkpoint, and rewinds the sources to the offsets recorded in that checkpoint.
// Recovery steps:
// 1. Detect failure (task manager crash)
// 2. Cancel all running tasks
// 3. Find last completed checkpoint (e.g., CP-41)
// 4. Restore every operator's state from CP-41
// 5. Rewind sources to offsets stored in CP-41
// 6. Resume processing from consistent point
// Result: exactly-once state semantics
Recovery timeline:
CP-40 ──▶ CP-41 ──▶ [processing...] ──▶ CRASH
▲
Last completed checkpoint
Restore all operators to CP-41 state
Rewind Kafka offsets to CP-41 positions
Resume processing
What about exactly-once end-to-end?
Checkpointing alone guarantees consistent state, not consistent external side effects. Exactly-once end-to-end additionally requires transactional or idempotent sinks.
// State exactly-once: guaranteed by checkpointing
// End-to-end exactly-once: requires transactional sink
// Example: two-phase commit Kafka sink
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(
"nbi-enriched-txns",
new SimpleStringSchema(),
properties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE // 2PC
);
stream.addSink(kafkaSink);
What are the performance considerations?
Barrier alignment can cause backpressure when one input is slow, because fast channels are buffered while waiting. Asynchronous snapshots and incremental checkpointing (with RocksDB) keep checkpoint overhead low.
// Monitor checkpoint metrics:
// - Checkpoint duration: time from trigger to completion
// - Alignment duration: time waiting for all barriers
// - State size: bytes uploaded per checkpoint
// - Checkpoints completed vs failed
// Configure timeouts
env.getCheckpointConfig().setCheckpointTimeout(60_000); // 60s timeout
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5_000); // 5s gap
What are common mistakes?
The most common mistake is not monitoring checkpoint duration. If checkpoints take longer than the interval, they overlap and eventually fail. Another mistake is forgetting that exactly-once state does not mean exactly-once side effects.
// WRONG -- checkpoint interval shorter than checkpoint duration
env.enableCheckpointing(5_000); // 5 seconds
// But state is 50 GB with full snapshots → takes 30 seconds
// Checkpoints overlap and fail!
// CORRECT -- use RocksDB incremental + adequate interval
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
env.enableCheckpointing(10_000); // 10 seconds
// Incremental: uploads only delta, completes in seconds
What should a senior engineer know?
Barrier alignment is what guarantees a consistent cut across the distributed graph. Asynchronous snapshots keep the pipeline moving. Unaligned checkpoints (Flink 1.11+) exist to solve the backpressure amplification problem of alignment. And exactly-once end-to-end requires transactional or idempotent sinks beyond just checkpointing.
Visual Overview
Flink Checkpoint Barrier Flow
┌──────────┐
│ JobMgr │ Trigger checkpoint 42
│Coordinator│
└────┬─────┘
│ inject barrier(42)
▼
┌──────────┐ ┌──────────┐
│ Source 1 │ │ Source 2 │
│ B42 ──▶ │ │ B42 ──▶ │
└────┬─────┘ └────┬─────┘
│ │
▼ ▼
┌────────────────────────────┐
│ Join Operator (2 inputs) │
│ │
│ Input 1: ─[R1][B42]──▶ │
│ Input 2: ─[C1][C2]──▶ │ ◀─ B42 not yet on Input 2
│ buffer Input 1 │ keep processing Input 2
│ │
│ Input 2: ─[C3][B42]──▶ │ ◀─ B42 arrives on both
│ SNAPSHOT state │
│ forward B42 │
└────────────┬───────────────┘
│
▼
┌──────────────────┐
│ Async Upload │
│ State → S3/HDFS │
│ ACK → Coord │
└──────────────────┘
Key Takeaway
Flink checkpointing is a distributed snapshot mechanism based on the Chandy-Lamport algorithm. The checkpoint coordinator in the JobManager periodically triggers a checkpoint by telling every source to inject a barrier marker into its output stream. The barrier flows downstream through the dataflow graph alongside normal records, never overtaking them, separating pre-checkpoint from post-checkpoint records.
The subtle part is barrier alignment for multi-input operators. When the barrier arrives on one input, the operator stops consuming that input and buffers its records while continuing the other inputs until their barriers arrive. Only when all barriers are present does the operator snapshot, guaranteeing the snapshot contains exactly all pre-barrier records. The snapshot is uploaded asynchronously to durable storage like S3, so the processing thread is barely blocked.
On failure, Flink restores every operator from the last completed checkpoint and rewinds sources to the stored offsets, resuming from a consistent point with exactly-once state semantics. Exactly-once end-to-end additionally requires transactional or idempotent sinks.
At the National Bank of India checkpointing was the backbone of the stateful running-balance job. Every ten seconds the coordinator triggered a checkpoint, barriers flowed through the merged retail and corporate streams, and each operator's per-account balance was snapshotted to S3. A task manager crash restored all state to one coherent instant, never losing or double-applying a transaction.
Remember one sentence: Barriers flow from sources through the graph, multi-input operators align on all barriers before snapshotting, and the async snapshot to durable storage produces one globally consistent recovery point for exactly-once state semantics.
---