Flink: Cannot Map Savepoint State for Operator — Fix It

1. The Error

You cancel a running job with a savepoint, deploy the new version of your jar, restore from that savepoint — and the JobManager refuses to start the job:

org.apache.flink.util.FlinkException: Could not restore from checkpoint/savepoint
    ...
Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint
s3://flink-state/savepoints/savepoint-2a1b3c-9f8e7d6c5b4a.
Cannot map checkpoint/savepoint state for operator 4f8b2c1d9e7a6b5c3d2e1f0a9b8c7d6e
to the new program, because the operator is not available in the new program.
If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
    at org.apache.flink.runtime.checkpoint.Checkpoints.throwNonRestoredStateException(Checkpoints.java:...)
    at org.apache.flink.runtime.checkpoint.Checkpoints.loadAndValidateCheckpoint(Checkpoints.java:...)

A close cousin appears from the assignment path when the topology changed but the operator ID still exists in the snapshot:

java.lang.IllegalStateException: There is no operator for the state 4f8b2c1d9e7a6b5c3d2e1f0a9b8c7d6e
    at org.apache.flink.runtime.checkpoint.StateAssignmentOperation.checkStateMappingCompleteness(...)

The operator ID in the message is a 128-bit hash, not your operator name — which is exactly why this is confusing. Nothing in your code is named 4f8b2c1d....

Applies to: Flink 1.14 through 2.2, all deployment modes (standalone, YARN, Flink Kubernetes Operator, application/session mode), any state backend (HashMapStateBackend or EmbeddedRocksDBStateBackend), canonical and native savepoints alike. The mechanism is version-independent; only a few config key names changed in 2.0 (noted below).

2. How to Reproduce It

You do not need Kafka or a cluster to see this — the operator-ID hashing is deterministic on the JobGraph. But a realistic reproduction uses a stateful job you upgrade.

docker-compose.yml for a local JM + TM:

services:
  jobmanager:
    image: flink:1.20.1-java17
    ports: ["8081:8081"]
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        state.savepoints.dir: file:///tmp/savepoints
        state.checkpoints.dir: file:///tmp/checkpoints
    volumes: ["./sp:/tmp/savepoints", "./cp:/tmp/checkpoints"]
  taskmanager:
    image: flink:1.20.1-java17
    depends_on: [jobmanager]
    command: taskmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 4
    volumes: ["./sp:/tmp/savepoints", "./cp:/tmp/checkpoints"]

Note: state.savepoints.dir is the Flink 1.x key (used above, since the image is 1.20.1). On Flink 2.x it was renamed to execution.checkpointing.savepoint-dir. Both set the default savepoint target directory.

Version 1 of the job — a keyed counter with no uid():

DataStream<Event> events = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "events");

events
    .keyBy(Event::userId)
    .process(new CountPerUser())   // stateful, but no uid()
    .sinkTo(sink);

Run it, then take a savepoint and stop:

# take a savepoint of a running job (job keeps running)
bin/flink savepoint <jobId> file:///tmp/savepoints

# or stop-with-savepoint (drains and stops cleanly)
bin/flink stop --savepointPath file:///tmp/savepoints <jobId>

Version 2 — you insert a trivial, stateless operator upstream (a common, innocent change: adding a filter or a metric map):

events
    .filter(e -> e.userId() != null)   // NEW operator inserted before keyBy
    .keyBy(Event::userId)
    .process(new CountPerUser())       // same logic, STILL no uid()
    .sinkTo(sink);

Restore:

bin/flink run -s file:///tmp/savepoints/savepoint-<id> ./job-v2.jar

It fails with Cannot map checkpoint/savepoint state for operator .... You changed nothing about the stateful operator's logic — but its auto-generated ID moved.

Environment-specific trigger: this fires only on restore from a snapshot and only when operator IDs were auto-generated. A job with stable uid() on every operator survives the same edit untouched.

3. Why It Happens — Surface Level

Every operator in a Flink job has an OperatorID. State in a savepoint is keyed by that ID, not by operator name or position. On restore, Flink matches each chunk of state in the savepoint to an operator in the new JobGraph by ID. If a savepoint ID has no matching operator in the new program, restore fails with Cannot map ... to the new program.

When you do not call uid(), Flink generates the ID by hashing the operator's position in the topology — its chained neighbours, inputs, and graph structure. Insert, remove, reorder, or re-chain an operator and the hash of downstream and neighbouring operators changes too, even if their business logic is byte-for-byte identical. The savepoint still carries the old IDs; the new graph advertises new ones; the intersection is empty for that operator, and restore aborts.

4. Why It Happens — Under the Hood

The ID assignment lives in the StreamGraphHasher (the StreamGraphHasherV2 implementation). During StreamGraph → JobGraph translation, Flink walks operators in a deterministic traversal and computes a hash per operator from: a per-operator user hash if uid() was set, otherwise the generated hash derived from the number of inputs, the hashes of its inputs, and chaining information. That hash becomes the OperatorID. This is why the ID is "sensitive to program changes" — the docs put it plainly: "The generated IDs depend on the structure of your program and are sensitive to program changes."

On restore, Checkpoints.loadAndValidateCheckpoint reads the savepoint metadata (the _metadata file), builds the set of operator IDs it contains, and cross-checks against the operator IDs in the submitted JobGraph. For any savepoint operator ID missing from the new job, it calls throwNonRestoredStateException unless allowNonRestoredState is set. Separately, once assignment begins, StateAssignmentOperation.checkStateMappingCompleteness enforces the reverse and the sizing: it verifies every piece of state has a home and that maxParallelism matches.

That last point ties this error to rescaling. maxParallelism (key-group count) is baked into the first checkpoint of an operator and cannot change afterward. If you also try to bump parallelism past the recorded max, you get a different failure from the same code path:

The maximum parallelism (128) with which the latest checkpoint of the execution
job vertex <id> has been taken and the current maximum parallelism (256) changed.
This is currently not supported.

So a botched upgrade can surface as either "no operator for this state" (topology drift) or a max-parallelism mismatch (rescale drift). They are the same subsystem — key-group-to-operator state assignment — failing at two different checks. (This is the sister problem to the checkpoint-lifecycle failures covered in the "Checkpoint expired before completing" article: same snapshot machinery, different failure surface.)

5. The Fix

Immediate, correct fix: give every stateful operator a stable uid(). Once the ID is pinned by you rather than derived from topology, upstream/downstream edits stop breaking it.

Before:

events
    .keyBy(Event::userId)
    .process(new CountPerUser())
    .sinkTo(sink);

After:

events
    .keyBy(Event::userId)
    .process(new CountPerUser())
    .uid("count-per-user")          // stable, human-chosen operator ID
    .name("count-per-user")         // display name (does NOT affect state mapping)
    .sinkTo(kafkaSink)
    .uid("events-sink");

Take a fresh savepoint from the currently-running old version (which now runs with uids), deploy, restore. From that point on, inserting a filter upstream is safe.

But you already have a savepoint from a job that had no uids. You cannot retroactively add uids and match the old auto-generated hashes by naming — the hash space is different. Two escape hatches:

1. setUidHash(...) — pin the operator to the old generated hash. Read the offending hash straight out of the error message (4f8b2c1d...) and assign it:

   events.keyBy(Event::userId)
       .process(new CountPerUser())
       .setUidHash("4f8b2c1d9e7a6b5c3d2e1f0a9b8c7d6e");  // migration bridge only

This makes the new operator adopt the savepoint's ID so its state maps. Use it once to bridge the upgrade, take a new savepoint, then replace it with a normal uid() and cut over. It is a migration tool, not a permanent annotation.

2. --allowNonRestoredState (-n) — skip the orphaned state. Only valid when you intend to drop that operator's state (you removed the operator, or the state is disposable):

   bin/flink run -s file:///tmp/savepoints/savepoint-<id> --allowNonRestoredState ./job-v2.jar

Danger: this silently discards any savepoint state that has no operator to land in. If you reach for it to "make the error go away" on an operator you actually kept (but whose hash drifted), you have just thrown away that operator's state and restarted it empty — a data-correctness bug, not a fix. Reserve it for genuinely-removed operators.

Which to use: setUidHash when you must preserve state through a one-time migration of an unversioned job; uid() for every job going forward; --allowNonRestoredState only when dropping state is the intended semantics.

6. Best Practices & The Better Design

The whole class of failure disappears if you treat operator IDs as a public contract. The right-way job:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Event> events = env
    .fromSource(kafkaSource, WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withIdleness(Duration.ofMinutes(1)),   // idle Kafka partitions won't stall watermarks
        "events")
    .uid("kafka-source-events");                // sources are stateful (offsets) — uid them too

events
    .filter(e -> e.userId() != null).uid("filter-null-user")
    .keyBy(Event::userId)
    .process(new CountPerUser()).uid("count-per-user")
    .sinkTo(kafkaSink).uid("events-sink");

Rules that make upgrades boring:

• uid() on every operator, including sources and sinks. Kafka sources hold offset state; transactional Kafka sinks hold pending-transaction state — both need stable IDs. (The sink's transactional state is where a missing uid meets ProducerFencedException after restore — a related failure worth its own note.)
• Never let uids be positional. Enforce it at the framework level:

  # config.yaml (Flink 2.x) — fail submission if any operator lacks a manual uid
  pipeline.auto-generated-uids: DISABLE

With this set, a job that forgot a uid() refuses to build — you catch it in CI, not in a 2 a.m. restore.

• Pin maxParallelism explicitly so a future rescale is not blocked by whatever value the first checkpoint happened to bake in:

  env.setMaxParallelism(32768);   // or a sane bound you plan to grow into

This connects uid stability to rescalability: stable uids let you change the graph; a well-chosen maxParallelism lets you change the parallelism. You need both to evolve a stateful job freely.

• Choose the savepoint format deliberately. Canonical is the default and is portable across state backends and Flink versions — use it for version upgrades and backend migrations. Native is faster to take/restore but backend-specific — fine for routine job restarts, not for cross-version moves.

7. How to Prevent It Long-Term

Make the operator-ID contract enforceable and observable:

• CI gate: compile the JobGraph in a unit test and assert every operator has a user-provided uid (or run with pipeline.auto-generated-uids: DISABLE in the test harness). This catches a missing uid before merge.
• Upgrade procedure, always: take a savepoint from the running job (bin/flink stop --savepointPath ... <jobId> for a clean drain, or bin/flink savepoint to snapshot without stopping), deploy the new jar, restore with bin/flink run -s. Never rely on the auto checkpoint for a planned upgrade.
• Keep an uid registry — a checked-in list of every operator uid per job. Reviewers reject PRs that rename or drop a uid without a migration plan.
• Monitor the restore, not just steady state: alert on numRestarts climbing right after a deploy and on the job sitting in INITIALIZING/RESTARTING. A restore that fails on state mapping shows up as a job that never reaches RUNNING.
• Chaos-test upgrades in staging: deploy version N+1 from a version-N savepoint on every release candidate. If someone dropped a uid, staging fails, not production.
• Reserve --allowNonRestoredState for reviewed, intentional state drops. Treat its appearance in a deploy script as a red flag requiring sign-off — it is the one flag that can destroy state without any error.

8. Key Takeaways

• Savepoint state maps to operators by OperatorID, not by name or position — and without uid(), that ID is a hash of your topology that moves whenever you edit the graph.
• Put uid() on every operator, sources and sinks included; it is the single change that makes stateful upgrades safe.
• To rescue a savepoint from an un-versioned job, bridge once with setUidHash("<hash-from-error>"), take a fresh savepoint, then switch to a normal uid().
• --allowNonRestoredState silently drops unmatched state — use it only when dropping that operator's state is what you actually mean.
• Enforce it structurally: pipeline.auto-generated-uids: DISABLE, a CI uid check, and a savepoint-based upgrade runbook turn this from a recurring outage into a non-event.