[FLINK-40113][checkpoint] Reject restore when operators on a keyed vertex disagree on max parallelism

[rionmonster]

What is the purpose of the change

When operators are chained into a single vertex, each operator's recorded maximum parallelism is reconciled onto that shared vertex individually during state restore. Since FLINK-31996 relaxed chaining to allow operators with differing maximum parallelism to share a vertex, a keyed operator can be silently restored under a key-group count that is not its own if the chaining topology has changed since the checkpoint — its keyed state is then remapped through an incompatible hash % maxParallelism layout.

Flink already rejects a single operator's incompatible maximum-parallelism change on restore, but never validates that operators sharing a vertex agree with each other, so the per-vertex reconciliation silently adopts whichever value is applied last. This change closes that consistency gap by rejecting such a restore with a clear error instead of restoring keyed state improperly.

Brief change log

• Add checkMaxParallelismAgreement in StateAssignmentOperation, invoked before the per-operator reconciliation in checkParallelismPreconditions.
• On a vertex that carries keyed state, fail with a clear IllegalStateException when its operators recorded differing maximum parallelism. The comparison spans all chained operators, since a non-keyed operator's recorded value can win the reconciliation and misroute a keyed operator's state.
• Add ChainingMaxParallelismStateLossITCase covering both directions (the downstream operator's explicit value below and above the chain head's) on the HashMap and RocksDB backends.
• Add StateAssignmentOperationTest#restoreRejectsKeyedVertexWithConflictingMaxParallelism, which drives the reconciliation directly and asserts rejection regardless of the order the operator states are reconciled.

Verifying this change

This change added tests and can be verified as follows:

• ChainingMaxParallelismStateLossITCase takes a canonical savepoint with operator chaining disabled, so a keyed operator (a keyBy chain head at an auto-derived maximum parallelism) and a downstream stateful operator (at a different explicit maximum parallelism) are separate vertices. It then restores with chaining enabled, so the downstream operator chains under the keyed head and the merged keyed vertex carries two differing recorded values, and asserts the restore is rejected with the new error for both directions (downstream value below and above the head's) and both state backends.
• StateAssignmentOperationTest#restoreRejectsKeyedVertexWithConflictingMaxParallelism drives the reconciliation directly and asserts the restore is rejected regardless of the order in which the operator states are reconciled onto the vertex.
• Existing StateAssignmentOperationTest and CheckpointCoordinatorRestoringTest continue to pass, confirming the new check is additive and runs only on restore.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — adds a restore-time precondition in checkpoint state assignment, executed once per vertex during recovery.
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? n/a

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.8 (1M context)

[flinkbot]

CI report:

• dd28e38 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[pnowojski]

LGTM % (assuming that both of the tests are failing without the fix as expected AND green azure)