Home / Data Platform / Streaming & CDC

Data PlatformRunbookAdvanced

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

Upgrade a stateful Flink job without losing state - pin operator UIDs, stop with a savepoint, check state compatibility, restore with the right claim mode and roll back if the new version misbehaves.

Deepak Bhardwaj6 min read

Applies toApache Flink 1.17 to 2.x, standalone, YARN and the Flink Kubernetes Operator; Kafka source and sink connectors

Goal

Deploy a new version of a stateful Flink job (code change, connector bump or Flink version upgrade) and resume from exactly where the old one stopped. No reprocessing, no duplicates from the sink and a tested way back.

The things that go wrong are predictable: auto-generated operator IDs changed, a state type was serialized with Kryo and can't evolve, max parallelism changed, or the restore silently dropped state with --allowNonRestoredState. This runbook checks for each one before the cutover.

Before the change window

1. Make sure every stateful operator has a UID

Flink maps savepoint state to operators by operator ID. Without .uid(), the ID is a hash of the topology, so inserting a map upstream changes it and the state no longer matches.

env.fromSource(kafkaSource, watermarks, "orders-source").uid("orders-source")
   .keyBy(Order::customerId)
   .process(new CustomerTotals()).uid("customer-totals").name("customer-totals")
   .sinkTo(kafkaSink).uid("totals-sink");

Enforce it so a missing UID fails at submit time instead of during the upgrade:

pipeline.auto-generate-uids: false

Warning

If the running job never had explicit UIDs, adding them now changes the IDs and the state will not map. Either set .setUidHash("<existing hash>") on each operator for this one upgrade (read the hashes from the savepoint's operator list or the web UI vertex IDs), or plan a state rebuild.

2. Check state type compatibility

State type Schema evolution
POJOs (Flink POJO rules) Add and remove fields; no type changes of existing fields
Avro (specific/generic) Avro compatibility rules
Kryo fallback (generic types) None in practice; changes often fail on restore
Key types Must not change

Find Kryo usage before it bites you:

pipeline.generic-types: false

With this set, the job fails at startup if any type falls back to Kryo. Run it in a test environment with the new build first.

3. Don't change max parallelism

Keyed state is split into key groups, and the number of key groups is the max parallelism. You can change parallelism up to that value on restore. You can't change max parallelism itself without rewriting the savepoint with the State Processor API. If it was never set explicitly, it was derived from the original parallelism. Set it explicitly now (setMaxParallelism) to the value the savepoint already uses.

4. Check where savepoints go

The savepoint target is a durable path (S3, ADLS, GCS, HDFS), not local disk. The config key is state.savepoints.dir in 1.x and execution.checkpointing.savepoint-dir in 2.x. Names vary by version, so check your build's docs. Confirm the job's identity has write access to it.

Cutover

5. Take a savepoint and stop the job

Stop-with-savepoint stops sources, takes a savepoint and commits pending sink transactions in one step:

flink list -r
flink stop --savepointPath s3://flink-state/savepoints/orders <jobId>

The command prints the savepoint path. Record it in the change ticket.

Caution

Don't add --drain for a normal upgrade. Draining emits a final MAX_WATERMARK, which fires all pending event-time windows and timers as if the stream had ended. The new job then has no open windows, and the results for in-flight windows are wrong. Use --drain only when you are retiring the pipeline.

With the REST API (useful from CI):

curl -s -X POST http://jobmanager:8081/jobs/$JOB/stop \
  -H 'Content-Type: application/json' \
  -d '{"targetDirectory":"s3://flink-state/savepoints/orders","drain":false}'
# returns a request-id; poll it
curl -s http://jobmanager:8081/jobs/$JOB/savepoints/$REQUEST_ID

If stop-with-savepoint times out (backpressure, a stuck sink), the job keeps running. Fix the cause first; don't fall back to flink cancel, which gives you only the last checkpoint.

6. Start the new version from the savepoint

flink run -d -s s3://flink-state/savepoints/orders/savepoint-ab12cd-0123456789ab \
  -p 8 orders-job-2.4.0.jar

Restore mode decides who owns the savepoint files:

Mode Behaviour Use when
NO_CLAIM (default) New job takes a full checkpoint first and never depends on the savepoint files afterwards Normal upgrades; keeps the savepoint valid for rollback
CLAIM New job takes ownership and may delete the files later Large state where the first full checkpoint is too expensive

The flag is --restoreMode in older releases and --claimMode in newer ones; check flink run --help on your version. Keep the default for upgrades so the rollback savepoint survives.

Warning

-n / --allowNonRestoredState lets the job start even if some savepoint state has no matching operator. It's correct when you deliberately removed an operator. Otherwise it hides a UID mismatch and starts that operator empty. Never put it in a deploy script by default.

With the operator, the same steps are driven by the FlinkDeployment spec:

spec:
  job:
    jarURI: local:///opt/flink/usrlib/orders-job-2.4.0.jar
    parallelism: 8
    upgradeMode: savepoint     # stateless | last-state | savepoint
    state: running

upgradeMode: savepoint makes the operator stop with a savepoint and restore from it when the spec changes. last-state uses the latest checkpoint and HA metadata. It's quicker, but it can't move between incompatible Flink versions. Check status.jobStatus and the savepoint info in the resource status after each change.

Validate

  • Web UI or GET /jobs/<id>/checkpoints: the first checkpoint completes and its size is similar to the old job's.
  • Restored state: per-operator state size in the checkpoint details isn't near zero for operators that should have state.
  • Kafka source: consumer lag drops from the savepoint offsets; it doesn't jump to latest or earliest. A jump means the source state didn't restore and the connector fell back to its configured starting offsets.
  • Kafka exactly-once sink: downstream read_committed consumers see data flowing. A changed transactionalIdPrefix between versions can leave old transactions open until transaction.timeout.ms expires, which stalls those consumers.
  • Business check: compare an aggregate (count per key for the last hour) against the old version's output.

Rollback

  1. flink stop --savepointPath ... the new version (or cancel it if it's unhealthy).
  2. Start the old jar from the pre-upgrade savepoint (step 5), with NO_CLAIM.
  3. Expect the old version to reprocess everything since that savepoint. With an exactly-once Kafka sink, duplicates are fenced. With idempotent sinks (upserts by key) they are harmless. With append-only sinks, plan for downstream dedup.

Tip

A savepoint taken by a newer Flink version usually can't be restored by an older one. Before a Flink version upgrade, the pre-upgrade savepoint is your only rollback point. Keep it until the new version has run cleanly for a while.

Prevent it next time

  • Fail builds on missing UIDs (pipeline.auto-generate-uids: false) and Kryo fallback (pipeline.generic-types: false).
  • Set max parallelism explicitly on day one.
  • Add a CI stage that restores the latest production savepoint (or a recent copy of it) into the new build and checks that it starts without --allowNonRestoredState.
  • Take savepoints in canonical format (flink savepoint --type canonical <jobId>) before Flink version upgrades, where the release notes require it.
  • Keep a small runbook entry per job: savepoint path, UIDs of stateful operators, max parallelism and sink transactional ID prefix.

Related fixes