Flink Checkpoint Expired Before Completing: Causes & Fixes

Your job runs fine for an hour, then the checkpoint counter in the Web UI stops advancing. A few minutes later the job restarts from an old checkpoint, reprocesses, and does it again — a slow, expensive loop. In the JobManager log you find this:

Checkpoint expired before completing.

This is the single most common checkpointing failure teams hit on their first real-traffic deploy. It almost never means "checkpointing is broken." It means a checkpoint could not finish inside execution.checkpointing.timeout, and once enough of them expire you cross tolerable-failed-checkpoints and the job fails over. This article covers exactly why it happens, how to reproduce it, and how to fix it properly — not by blindly raising the timeout.

1. The Error

The canonical form, logged by the CheckpointCoordinator on the JobManager:

org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Failed to trigger or complete checkpoint 128 for job 3f1c... (0 bytes, checkpointDuration=600021 ms).
org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint tolerable failure threshold.

And the decline reason, which is the string everyone searches for:

Checkpoint 128 of job 3f1c... expired before completing.

When the threshold is crossed you also get the job-killing exception:

org.apache.flink.runtime.checkpoint.CheckpointException: Checkpoint expired before completing.
	at org.apache.flink.runtime.checkpoint.CheckpointCoordinator...
Caused by: ... Exceeded checkpoint tolerable failure threshold.

In the Web UI, the Checkpoints → History tab shows the checkpoint stuck in IN_PROGRESS, then flipping to FAILED at exactly the timeout value, with one or more subtasks showing an "n/a" or partial Acknowledged count.

Environment this article targets: Apache Flink 1.18–1.20 (config keys noted where they were renamed), DataStream API and Flink SQL, standalone / Kubernetes application mode / Flink Kubernetes Operator, RocksDB or HashMap state backend, checkpoints on S3/GCS/HDFS. The CheckpointFailureReason behind the log line is CHECKPOINT_EXPIRED.

2. How to Reproduce It

The fastest reliable reproduction is a slow sink plus a short timeout. Below is a minimal local cluster and job.

docker-compose.yml:

services:
  jobmanager:
    image: flink:1.20.0-java17
    ports: ["8081:8081"]
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        state.backend.type: rocksdb
        execution.checkpointing.interval: 10s
        execution.checkpointing.timeout: 30s
        execution.checkpointing.tolerable-failed-checkpoints: 2
  taskmanager:
    image: flink:1.20.0-java17
    depends_on: [jobmanager]
    command: taskmanager
    scale: 1
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 4

A job that guarantees backpressure by sleeping in the sink while a fast source keeps producing:

public class SlowSinkJob {
  public static void main(String[] args) throws Exception {
    var env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(4);

    env.fromSequence(1, Long.MAX_VALUE)          // fast, unbounded source
       .keyBy(v -> v % 8)
       .process(new StatefulCounter())           // builds RocksDB state
       .addSink(new RichSinkFunction<Long>() {
          @Override public void invoke(Long value, Context ctx) throws Exception {
            Thread.sleep(50);                      // slow sink -> backpressure
          }
       });

    env.execute("slow-sink-checkpoint-expiry");
  }
}

Submit and watch:

flink run -d target/slow-sink-job.jar
flink list
# Poll checkpoint stats via REST:
curl -s localhost:8081/jobs/<jobid>/checkpoints | jq '.latest.failed'

Within a minute the checkpoint duration climbs past 30s, latest.failed populates with "failure_message": "Checkpoint expired before completing.", and after the third expiry the job fails over.

Environment-specific triggers. This surfaces only under backpressure, or only with RocksDB and large state (slow async snapshot upload), or only on S3 checkpoint storage (throttled/slow PUTs), or only after a rescale when one slow subtask holds up the whole checkpoint. On a healthy, unbackpressured job with small state you will never see it.

3. Why It Happens — Surface Level

A checkpoint has a deadline. execution.checkpointing.timeout (default 10 min, method setCheckpointTimeout()) is the wall-clock budget from the moment the CheckpointCoordinator triggers checkpoint N to the moment every subtask has acknowledged it. If any subtask fails to acknowledge in time, the coordinator declines the checkpoint with CHECKPOINT_EXPIRED and logs "expired before completing."

By default execution.checkpointing.tolerable-failed-checkpoints is 0, so historically a single expiry could fail the job. More importantly, once the number of consecutive failed checkpoints exceeds whatever you set, the coordinator throws "Exceeded checkpoint tolerable failure threshold" and the job restarts. The expiry is the symptom; the real question is why one or more subtasks couldn't finish in time.

4. Why It Happens — Under the Hood

To finish a checkpoint, a barrier injected at the sources has to flow all the way to the sinks, and every operator has to snapshot its state along the way. Two phases dominate the wall clock, and either can blow the budget.

Phase 1 — barrier alignment (the usual culprit). With aligned checkpoints (the default, exactly-once), an operator with multiple input channels must wait until the barrier arrives on all of them before it snapshots and forwards the barrier. Under backpressure, the barrier travels only as fast as the data in the buffers ahead of it. If a downstream operator is slow (our sleeping sink), buffers fill, credit-based flow control stops granting credits upstream, and the barrier sits behind a wall of buffered records. The "checkpoint start delay" — the time from trigger to when the operator even sees the barrier — balloons. This alignment time is pure dead time counted against your timeout. The metric to watch is checkpointStartDelayNanos; when it approaches your timeout, alignment is your problem.

Phase 2 — async state snapshot. After alignment, each operator snapshots. Flink splits this into a fast synchronous part (flush memtables / take a RocksDB snapshot handle) and an asynchronous part (upload state files to durable storage). For large RocksDB state on S3, the async upload is where time goes — and incremental checkpoints reduce bytes but can multiply file count, and thousands of small PUTs to a throttled bucket are slow. A single overloaded TaskManager, a GC pause during the sync phase, or a slow DFS endpoint can push one subtask past the deadline while every other subtask acknowledged in milliseconds.

The CheckpointCoordinator doesn't care which phase was slow — it only sees that subtask acknowledgements didn't all arrive before timeout. That's why raising the timeout "fixes" it temporarily: you're giving a slow alignment or a slow upload more room, without addressing the slowness. This is also why the failure is tightly coupled to backpressure and to your Kafka transactional sink — the same barrier that stalls here is the one that must reach the sink for it to commit its transaction, so a checkpoint that never completes means a sink that never commits (see the interlink note in §6).

5. The Fix

Diagnose first: open Checkpoints → History → the failed checkpoint and read per-subtask Sync Duration, Async Duration, and Alignment Duration / Start Delay. High Start Delay ⇒ backpressure/alignment problem. High Async Duration ⇒ state size / storage problem.

Fix A — kill alignment stalls with unaligned checkpoints (backpressure case). Unaligned checkpoints let the barrier overtake buffered in-flight data and snapshot it as part of state, so alignment time collapses toward zero. This is the correct fix when Start Delay dominates.

# BEFORE — aligned, exactly-once; barrier stuck behind buffers under backpressure
execution.checkpointing.timeout: 10min
execution.checkpointing.tolerable-failed-checkpoints: 0

# AFTER — start aligned, auto-switch to unaligned if alignment drags past 30s
execution.checkpointing.unaligned.enabled: true     # older key: execution.checkpointing.unaligned
execution.checkpointing.aligned-checkpoint-timeout: 30s  # older key: execution.checkpointing.alignment-timeout
execution.checkpointing.tolerable-failed-checkpoints: 3

The aligned-checkpoint-timeout (setter setAlignedCheckpointTimeout(Duration)) means "stay cheap and aligned normally, but if a checkpoint is clearly stuck aligning, flip it to unaligned so it still completes." Costs: unaligned checkpoints store in-flight buffers, so checkpoint size grows, and they don't support concurrent checkpoints > 1 or certain rescaling paths. Use them when backpressure is real, not as a blanket default.

Fix B — shrink async snapshot time (large-state case).

# Enable incremental checkpoints so you upload deltas, not full state
state.backend.incremental: true
# Give RocksDB enough managed memory so it isn't stalling on flushes/compaction
taskmanager.memory.managed.fraction: 0.5
# Speed up recovery too
state.backend.local-recovery: true

Fix C — the honest short-term lever. Raising the timeout is legitimate when the job genuinely needs more time (very large state on slow storage) and you've confirmed it isn't a runaway backpressure loop:

execution.checkpointing.timeout: 30min

Do not reach for this first. A large timeout on a backpressured job just means each failed attempt wastes 30 minutes before failing.

Choosing between them: Start Delay high → Fix A. Async Duration high → Fix B. Both healthy but state is legitimately huge → Fix C.

6. Best Practices & The Better Design

The better design removes the pressure that makes checkpoints slow in the first place, rather than tuning the deadline.

# --- config.yaml: a sane production checkpoint profile ---
execution.checkpointing.interval: 60s
execution.checkpointing.min-pause: 30s          # always leave breathing room between checkpoints
execution.checkpointing.timeout: 10min
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.tolerable-failed-checkpoints: 5   # survive transient blips, don't restart-storm
execution.checkpointing.unaligned.enabled: true
execution.checkpointing.aligned-checkpoint-timeout: 30s
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

# Buffer debloating: shrink in-flight data so alignment is fast even when aligned
taskmanager.network.memory.buffer-debloat.enabled: true

state.backend.type: rocksdb
state.backend.incremental: true
state.backend.local-recovery: true

Three principles behind this:

• min-pause is your friend. Setting min-pause (30s here) guarantees a gap between the end of one checkpoint and the start of the next, so a job that's briefly slow doesn't queue back-to-back checkpoints and dig itself deeper. It effectively decouples "how often I try" from "how often I succeed."
• Buffer debloating dynamically sizes network buffers to ~1s of data, so even aligned checkpoints align fast — often removing the need for unaligned entirely.
• Fix the backpressure, not the checkpoint. If a sink is the bottleneck (slow JDBC, throttled external API), scale it, batch it, or async it. A checkpoint that expires is usually Flink telling you a specific operator can't keep up — the same signal you'd otherwise see as consumer lag.

Interlink note: this connects directly to the Kafka transactional sink — with read_committed downstream consumers, records only become visible when the sink commits its transaction at checkpoint completion, so a stalling checkpoint and a transaction.timeout.ms misconfiguration are two sides of the same coin. It also connects to restart strategies: a low tolerable-failed-checkpoints plus an aggressive restart strategy produces the restart-storm loop described in §1.

7. How to Prevent It Long-Term

Alert on the metrics that predict this failure before it fails the job:

• lastCheckpointDuration trending toward execution.checkpointing.timeout — leading indicator.
• numberOfFailedCheckpoints / numberOfInProgressCheckpoints — a rising failed count is the warning shot.
• checkpointStartDelayNanos — high and rising means alignment/backpressure.
• isBackPressured and busyTimeMsPerSecond per operator — find which operator is the bottleneck.
• numRestarts — catches the loop early.
• RocksDB state.backend.rocksdb.estimate-live-data-size — state growth outrunning your timeout.

Team conventions that prevent it: set tolerable-failed-checkpoints to a small positive number (3–5) in every job so a transient blip doesn't fail over; standardize min-pause so nobody ships an interval-only config; capacity-plan so no single subtask is a permanent bottleneck; and chaos-test by killing a TaskManager pod under load to confirm checkpoints still complete during recovery. In CI, assert that the effective checkpoint config includes a non-zero min-pause and tolerable-failed-checkpoints before promoting to production.

8. Key Takeaways

• "Checkpoint expired before completing" = a subtask didn't acknowledge within execution.checkpointing.timeout (default 10 min); the job dies once you exceed tolerable-failed-checkpoints (default 0).
• Diagnose by phase: high Start Delay ⇒ backpressure/alignment; high Async Duration ⇒ state size / slow storage. The Web UI checkpoint details tell you which.
• Backpressure case → unaligned checkpoints with an aligned-checkpoint-timeout; large-state case → incremental checkpoints + enough managed memory. Raising the timeout is a last resort, not a first move.
• min-pause + buffer debloating prevent the whole class of failure; a small positive tolerable-failed-checkpoints prevents transient blips from triggering restart storms.
• A checkpoint that won't complete is the same backpressure signal as consumer lag — and it blocks your Kafka transactional sink from committing. Fix the bottleneck, not just the deadline.