Flink checkpoints failing or timing out: backpressure, slow state uploads and unaligned checkpoints

Symptom

The job keeps running, but checkpoints don't complete. You see one or more of these:

The Checkpoints tab shows a run of FAILED entries with "Checkpoint expired before completing".
The job restarts with "Exceeded checkpoint tolerable failure threshold".
Checkpoint duration creeps from seconds to minutes, and Kafka consumer lag grows because offsets are only committed on completed checkpoints.
After a restart the job restores from a checkpoint that is hours old and reprocesses a large backlog.

A failing checkpoint is a recovery problem waiting to happen: if the job dies now, it rolls back to the last good one. Treat a long run of failures as urgent even if output looks normal.

Checks

1. Read the checkpoint statistics

The REST API gives you the same numbers as the UI, which is easier to paste into a ticket:

FLINK=http://jobmanager:8081
JOB=$(curl -s $FLINK/jobs | jq -r '.jobs[] | select(.status=="RUNNING") | .id' | head -1)

# Counts, latest completed/failed checkpoint and failure message
curl -s $FLINK/jobs/$JOB/checkpoints | jq '{counts, latest: {completed: .latest.completed | {id, end_to_end_duration, state_size}, failed: .latest.failed | {id, failure_message}}}'

# Interval, timeout, min pause, unaligned on/off
curl -s $FLINK/jobs/$JOB/checkpoints/config | jq

Then open one failed checkpoint in the UI and expand it per operator and per subtask. Two columns tell you where the time goes:

Column	What it measures	Large value means
Start delay (checkpoint start delay)	Time for the barrier to reach the subtask	Barriers stuck behind backpressured buffers
Alignment duration	Time waiting for barriers on all inputs	One slow input channel holding the others
Sync duration	Snapshotting state in the task thread	Large heap state, slow `snapshotState()` code
Async duration	Uploading state to the checkpoint store	Slow or throttled S3/ADLS/HDFS, huge state deltas

If only one or two subtasks are slow, suspect data skew or a single bad TaskManager rather than global settings.

2. Check for backpressure

# List vertices, then check backpressure per vertex
curl -s $FLINK/jobs/$JOB | jq -r '.vertices[] | "\(.id)  \(.name)"'
curl -s $FLINK/jobs/$JOB/vertices/<vertex-id>/backpressure | jq '{status, backpressureLevel, subtasks: [.subtasks[] | {subtask, backpressureLevel, ratio}]}'

The useful task metrics are backPressuredTimeMsPerSecond, busyTimeMsPerSecond and idleTimeMsPerSecond. The bottleneck is usually the first operator downstream of the backpressured ones that shows high busy time (close to 1000 ms/s). Everything upstream of it is backpressured; everything after it is idle.

3. Check the checkpoint store

Look at the TaskManager logs around the failure for upload errors and throttling (S3 SlowDown, ADLS 503, HDFS timeouts), and at the size trend:

curl -s $FLINK/jobs/$JOB/checkpoints | jq '.history[] | {id, status, state_size, checkpointed_size, end_to_end_duration}'

checkpointed_size is what was uploaded for that checkpoint (with incremental RocksDB checkpoints this is the delta). A delta that keeps growing points at state that never expires.

Cause → fix

What you see	Likely cause	Fix
High start delay, operators backpressured	Barriers queued behind full network buffers	Fix the bottleneck operator; enable buffer debloating or unaligned checkpoints
High alignment on joins/unions	One input much slower than the other	Unaligned checkpoints; rebalance the slow input
High async duration, small backpressure	Slow store or big uploads	Incremental checkpoints, faster storage, check throttling
High sync duration	Heap state backend with large state, or heavy custom snapshot code	Move to RocksDB; review `CheckpointedFunction` code
State size grows forever	No TTL, missing watermarks, unbounded keys	State TTL, fix watermarks, bound key space
Fails at a fixed interval, then restarts	Timeout too short for current state size	Raise timeout and min pause while you fix the root cause
One subtask always last	Key skew or a sick TaskManager	Fix skew; restart or replace that TaskManager

Fixes

Buy time first

Raise the timeout and give the job room between checkpoints so it isn't permanently checkpointing. In config.yaml (or flink-conf.yaml on older versions):

execution.checkpointing.interval: 2 min
execution.checkpointing.timeout: 20 min
execution.checkpointing.min-pause: 1 min
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.tolerable-failed-checkpoints: 5

Applying cluster config changes means restarting the job. Restart from a savepoint or the latest retained checkpoint, not from empty state:

flink stop --savepointPath s3://flink-savepoints/orders $JOB
flink run -s s3://flink-savepoints/orders/savepoint-xxxx -d orders-job.jar

Warning
If checkpoints are failing because of backpressure, a stop-with-savepoint can time out for the same reason. If it does, the job keeps running. Don't fall back to flink cancel until you know which retained checkpoint you'll restore from, or you can lose state and replay hours of data.

Remove the backpressure

Raising timeouts only hides the problem. The lasting fix is making the slow operator keep up: more parallelism for that operator, async I/O instead of blocking calls to external systems, batching sink writes, or fixing key skew. If the sink is the bottleneck (a database, Elasticsearch, a rate-limited API), checkpoints will stay slow until the sink is faster.

Enable buffer debloating

Buffer debloating shrinks in-flight network buffers to what the job can process in about one second, so aligned barriers have less data to wait behind:

taskmanager.network.memory.buffer-debloat.enabled: true

It helps most on jobs with high parallelism and a steady backlog. It doesn't help if a single subtask is blocked.

Use unaligned checkpoints

Unaligned checkpoints let barriers overtake in-flight data, and store that data as part of the checkpoint. Checkpoint time stops depending on how backed up the pipeline is.

execution.checkpointing.unaligned.enabled: true   # older 1.x builds: execution.checkpointing.unaligned
execution.checkpointing.aligned-checkpoint-timeout: 30 s  # start aligned, switch to unaligned if barriers stall

Option names vary by version (Flink 2.0 reorganised many checkpointing options); check the configuration page for your build. Trade-offs to know about:

Checkpoints get larger and restores take longer, because in-flight buffers are stored too.
They only apply to checkpoints. Savepoints are always aligned, so a savepoint under heavy backpressure can still time out.
Check the docs for your version on limitations with some partitioners and when changing parallelism on restore.

Make state uploads cheaper

For RocksDB, turn on incremental checkpoints so each checkpoint uploads only new SST files:

state.backend.type: rocksdb
execution.checkpointing.incremental: true   # older builds: state.backend.incremental
execution.checkpointing.dir: s3://flink-checkpoints/orders  # older builds: state.checkpoints.dir

Then stop state from growing without bound. Set TTL on keyed state that doesn't need to live forever:

StateTtlConfig ttl = StateTtlConfig
    .newBuilder(Duration.ofHours(24))   // older APIs take Time.hours(24)
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .cleanupInRocksdbCompactFilter(1000)
    .build();
descriptor.enableTimeToLive(ttl);

Caution
Adding TTL to existing state changes its serialized format. Restoring old state into an operator that now has TTL enabled (or the reverse) can fail. Test the restore from a production savepoint on a copy of the job before rolling it out.

For SQL jobs, table.exec.state.ttl sets a default idle-state TTL for joins and aggregations. Rows whose state expires produce different results if late data arrives, so agree the value with the data owners.

Prevent it next time

Alert on checkpoint health, not just job status: numberOfFailedCheckpoints increasing, lastCheckpointDuration above half the timeout, and lastCheckpointSize growth week on week.
Keep a dashboard of backPressuredTimeMsPerSecond and busyTimeMsPerSecond per operator so you can find the bottleneck in minutes.
Retain checkpoints on cancellation (execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION; names vary by version) so you always have a restore point.
Load-test new versions with production-sized state and a replayed backlog, and compare checkpoint durations before release.
Set TTLs and watermarks when you design the job, not after state reaches hundreds of GB.

Flink checkpoints failing or timing out: backpressure, slow state uploads and unaligned checkpoints

#Symptom

#Checks

#1. Read the checkpoint statistics

#2. Check for backpressure

#3. Check the checkpoint store

#Cause → fix

#Fixes

#Buy time first

#Remove the backpressure

#Enable buffer debloating

#Use unaligned checkpoints

#Make state uploads cheaper

#Prevent it next time

Flink job upgrade runbook - stop with savepoint, keep operator UIDs stable and restore state safely

Flink event-time checklist: stalled watermarks, idle Kafka partitions, late data and watermark alignment

Kafka Streams stuck rebalancing or restoring state for hours - changelogs, standby replicas and store names

Symptom

Checks

1. Read the checkpoint statistics

2. Check for backpressure

3. Check the checkpoint store

Cause → fix

Fixes

Buy time first

Remove the backpressure

Enable buffer debloating

Use unaligned checkpoints

Make state uploads cheaper

Prevent it next time