Phase 4: Fault Tolerance, Checkpoints & Exactly-Once Semantics¶

Audience: Principal Distributed Systems Architect & Fault Tolerance Specialist
Source References: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/, flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java

1. Asynchronous Barrier Snapshotting (ABS Algorithm)¶

Flink achieves distributed fault tolerance using an optimized variant of the Chandy-Lamport Distributed Snapshot Algorithm:

ABS Checkpoint Execution Flow:
1. CheckpointCoordinator (JobManager) periodically generates CheckpointID (e.g. 501).
2. Coordinator sends RPC to all Source operators: TriggerCheckpoint(501).
3. Sources capture their current read offsets (e.g. Kafka partition offsets).
4. Sources inject CheckpointBarrier(501) into output data streams.
5. Barriers flow downstream interleaved with data records.

Stream Flowing Through Operator:
[Record A] ──► [Record B] ──► [CheckpointBarrier 501] ──► [Record C] ──► [Record D]
                                       │
                                       ▼
                     Delineates State: Records before Barrier
                     are included in Checkpoint 501. Records after
                     Barrier belong to Checkpoint 502!

2. Aligned vs. Unaligned Checkpoints¶

┌────────────────────────────────────────────────────────────────────────┐
│ 1. Aligned Checkpoints (Default)                                       │
├────────────────────────────────────────────────────────────────────────┤
│ - Operator waits until Barrier N arrives on ALL parallel input channels│
│ - Records arriving on Channel 1 are buffered in RAM while waiting for  │
│   Channel 2's barrier.                                                 │
│ - Flaw: Under high backpressure, Barrier N is stuck behind thousands   │
│   of records in network buffers, causing alignment timeout failures.   │
└────────────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────────────┐
│ 2. Unaligned Checkpoints (High-Throughput / Backpressure Resilient)    │
├────────────────────────────────────────────────────────────────────────┤
│ - The instant Barrier N arrives on ANY channel, it JUMPS the queue!    │
│ - Operator snapshots its internal state PLUS all in-flight buffers!    │
│ - Zero alignment wait time! Completes in seconds under 100% backpressure│
└────────────────────────────────────────────────────────────────────────┘

Production Configuration:¶

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);

CheckpointConfig config = env.getCheckpointConfig();
config.enableUnalignedCheckpoints(); // Jump network queues under backpressure
config.setAlignmentTimeout(Duration.ofSeconds(2)); // Try aligned for 2s, then switch!
config.setMinPauseBetweenCheckpoints(5000); // Prevent checkpoint storm starvation
config.setCheckpointTimeout(60000);

3. End-to-End Exactly-Once with Two-Phase Commit (2PC) Sinks¶

Generating correct internal state snapshots is useless if the external storage sink (e.g. Kafka, PostgreSQL) writes duplicate records on recovery.

To achieve True End-to-End Exactly-Once, Flink integrates TwoPhaseCommitSinkFunction:

sequenceDiagram
    autonumber
    participant App as "Flink Worker Task"
    participant Sink as "TwoPhaseCommitSink (Kafka Sink)"
    participant Kafka as "Kafka Broker (Transactional Coordinator)"
    participant Coord as "JobManager (CheckpointCoordinator)"

    Note over App,Sink: Normal Streaming Phase
    App->>Sink: Process records
    Sink->>Kafka: Write records under Transaction T1

    Note over Coord,Sink: Phase 1: Pre-Commit (Checkpoint Triggered)
    Coord->>App: Deliver CheckpointBarrier(100)
    App->>Sink: Forward Barrier(100)
    Sink->>Kafka: Flush in-flight records and Pre-commit Transaction T1
    Sink->>Sink: Open new Transaction T2 for incoming records!
    Sink->>Coord: Acknowledge Checkpoint 100 Pre-Committed

    Note over Coord,Sink: Phase 2: Commit (Global Checkpoint Committed)
    Coord->>Sink: NotifyCheckpointComplete(100)
    Sink->>Kafka: Commit Transaction T1! (Records visible to consumers!)

3.1 What Happens on Failure Recovery?¶

If a TaskManager crashes before Step 8: 1. JobManager restarts the job from Checkpoint 99. 2. The uncommitted Transaction T1 in Kafka times out and is aborted by the broker. 3. Flink replays the exact stream from Checkpoint 99's offsets. 4. Result: Zero duplicates, zero data loss, exact mathematical fidelity!