Aligned vs unaligned checkpoints in Flink

• Aligned checkpoint – Operator waits for the barrier on all input channels before snapshotting, buffering fast channels; produces small, clean snapshots but amplifies backpressure when one input is slow.
• Unaligned checkpoint – Operator snapshots immediately when the first barrier arrives, including in-flight records from input and output buffers; dramatically shortens checkpoint duration under backpressure.
• In-flight data – Records sitting in network buffers that are included in the unaligned snapshot and replayed on recovery.
• Alignment timeout – Hybrid setting where Flink starts aligned and switches to unaligned only if alignment takes too long.
• Trade-off – Unaligned checkpoints produce larger snapshots and heavier recovery but keep checkpoints reliable when the system is stressed.

How does aligned checkpointing work?

In aligned checkpointing, when a barrier arrives on one input channel of a multi-input operator, the operator stops consuming that channel and buffers further records while continuing others. Only when all channels have delivered the barrier does the operator snapshot.

// Aligned checkpointing (default)
StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(10_000);
env.getCheckpointConfig().setCheckpointingMode(
    CheckpointingMode.EXACTLY_ONCE);  // uses aligned by default

Aligned checkpoint with 2 inputs:
  Input 1 (fast): [R1][R2][B42][R3][R4]  ← barrier arrives first
  Input 2 (slow): [C1]...........[B42]   ← barrier arrives late

  Step 1: B42 on Input 1 → stop Input 1, buffer R3, R4
  Step 2: Wait for B42 on Input 2 (slow!)
  Step 3: B42 on Input 2 → snapshot state
  Step 4: Forward B42, drain buffered R3, R4

  Problem: Input 1 was stalled for the entire wait!
  Backpressure propagates upstream on Input 1.

What is the backpressure amplification problem?

When one input is slow (due to skew or a slow upstream), the operator buffers the fast channel's records while waiting, which propagates backpressure upstream on the fast channel. Under heavy backpressure, aligned checkpoints can take very long or time out entirely.

// Scenario: market-open spike
// Input 1 (transactions): 100K records/sec
// Input 2 (reference data): 1K records/sec (slow)

// Aligned CP: barrier on Input 1 at t=0
// Wait for Input 2's barrier... takes 30 seconds
// During those 30 seconds, Input 1 is buffered
// 3 million records buffered! Memory pressure.
// Backpressure sent upstream on Input 1.
// Checkpoint may time out.

How does unaligned checkpointing work?

With unaligned checkpoints (Flink 1.11+), the operator does not wait. When the first barrier arrives on any input, it immediately snapshots its state plus all in-flight records in its input and output buffers. The barrier can overtake buffered records.

// Enable unaligned checkpoints
env.enableCheckpointing(10_000);
env.getCheckpointConfig().enableUnalignedCheckpoints();

// Or with alignment timeout (hybrid approach)
env.getCheckpointConfig().setAlignedCheckpointTimeout(
    Duration.ofSeconds(5));  // start aligned, switch after 5s

Unaligned checkpoint with 2 inputs:
  Input 1 (fast): [R1][R2][B42][R3][R4]
  Input 2 (slow): [C1][C2]...[C3][C4]

  Step 1: B42 on Input 1 → SNAPSHOT IMMEDIATELY
    Snapshot includes:
    - Operator state
    - In-flight records from Input 2 buffer: [C1, C2]
    - In-flight records from output buffer
  Step 2: Forward B42 downstream (overtakes C1, C2)
  Step 3: Continue processing all inputs (no stall!)

  Result: checkpoint completes in milliseconds, no backpressure
  Cost: larger snapshot (includes in-flight data)

How are they different?

| Aspect | Aligned | Unaligned |
|---|---|---|
| Wait for all barriers | Yes | No |
| In-flight data in snapshot | No | Yes |
| Snapshot size | Smaller | Larger |
| Backpressure behavior | Amplifies backpressure | No additional backpressure |
| Checkpoint speed under load | Slow or timeout | Fast |
| Recovery cost | Lower (no in-flight replay) | Higher (replay in-flight) |
| Best for | Normal load, no backpressure | Backpressured pipelines |

// Aligned: small snapshot, slow under backpressure
// Unaligned: large snapshot, fast under backpressure

What is the alignment timeout hybrid?

The practical sweet spot is a hybrid: start aligned and automatically switch to unaligned if alignment takes too long. This keeps snapshots small during normal operation and pays the larger-snapshot cost only when needed.

// Hybrid: aligned by default, unaligned after 5 seconds
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setAlignedCheckpointTimeout(
    Duration.ofSeconds(5));

// Normal operation: alignment completes in < 5s → small aligned snapshot
// Under backpressure: alignment takes > 5s → switches to unaligned
//   → snapshot includes in-flight data → completes fast

What are the restrictions?

Unaligned checkpoints require exactly-once mode and have some restrictions depending on the Flink version. They may not combine with certain features, so check the release notes.

// Unaligned checkpoints require EXACTLY_ONCE mode
env.getCheckpointConfig().setCheckpointingMode(
    CheckpointingMode.EXACTLY_ONCE);  // Required
env.getCheckpointConfig().enableUnalignedCheckpoints();

// AT_LEAST_ONCE mode does NOT support unaligned checkpoints
// env.getCheckpointConfig().setCheckpointingMode(
//     CheckpointingMode.AT_LEAST_ONCE);
// env.getCheckpointConfig().enableUnalignedCheckpoints(); // ERROR

What are common mistakes?

The most common mistake is not enabling unaligned checkpoints when the job experiences backpressure and aligned checkpoints are timing out. Another is enabling unaligned checkpoints unnecessarily, adding snapshot size overhead when the job is not backpressured.

// WRONG -- aligned checkpoints timing out under backpressure
env.enableCheckpointing(10_000);
// Checkpoints fail repeatedly → no recovery point → dangerous!

// CORRECT -- enable unaligned with alignment timeout
env.enableCheckpointing(10_000);
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setAlignedCheckpointTimeout(
    Duration.ofSeconds(5));

What should a senior engineer know?

Unaligned checkpoints trade larger state and heavier recovery for reliable, fast checkpointing under load. The alignment-timeout hybrid is the real-world default because it keeps snapshots small in normal operation and only pays the unaligned cost when the system is struggling. Monitor alignment buffering time as a metric to detect when unaligned is kicking in.