[FLINK-40113][checkpoint] Reject restore when operators on a keyed vertex disagree on max parallelism#28714
[FLINK-40113][checkpoint] Reject restore when operators on a keyed vertex disagree on max parallelism#28714rionmonster wants to merge 1 commit into
Conversation
| // reinterpretAsKeyedStream puts a forward (chainable) edge before the keyed operator, so it | ||
| // can become a chained non-head under the source head. | ||
| final KeyedStream<Integer, Integer> keyed = | ||
| DataStreamUtils.reinterpretAsKeyedStream( | ||
| source, (KeySelector<Integer, Integer>) value -> value % NUM_KEYS); |
There was a problem hiding this comment.
Can this issue be reproduced without reinterpretAsKeyedStream? 🤔 Just normal keyBy followed by a chained/not chained operator?
reinterpretAsKeyedStream is a bit hacky and doesn't represents the normal use cases. It's intention is if partitions in your sources are already keyed by something, so you don't need to keyBy that data stream again. Here it only works because you constrain the parallelism to 1 which is also a bit artificial constraint.
There was a problem hiding this comment.
Yes, I agree and don't typically don't prefer using it.
I went down that road and tried iterating several times with just a plain keyBy to trigger the issue with very no luck. Per Claude:
keyBy inserts a hash (KeyGroupStreamPartitioner) edge, which breaks the chain, so the keyed operator always ends up as a chain head with its own maxParallelism. There's no second operator on the vertex to disagree with, and on restore it just adopts its own recorded value. To hit this the keyed operator has to be a chained non-head sharing a vertex with something that recorded a different maxParallelism, and reinterpretAsKeyedStream (forward edge) is the only pure-streaming way to get that topology.
I'll try exploring a few other ways to reproduce this issue without it and report back. 👍
There was a problem hiding this comment.
Why can't the non-keyed operator be chained to AFTER the keyed operator? Why doesn't that trigger the bug?
…rtex disagree on max parallelism During restore, each operator's recorded max parallelism is reconciled onto its shared vertex individually. When chaining has regrouped operators since the checkpoint, they can disagree, and the reconciliation silently keeps whichever value is applied last -- restoring a keyed operator under a foreign key-group count. Before the loop, verify that all operators chained into a vertex carrying keyed state agree, and fail with a clear error otherwise. The check spans all operators, since a non-keyed one's value can win the reconciliation and misroute a keyed operator's state. Adds ChainingMaxParallelismStateLossITCase covering both directions (explicit value below and above the chain head) on the HashMap and RocksDB backends.
3 participants
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 % maxParallelismlayout.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
checkMaxParallelismAgreementinStateAssignmentOperation, invoked before the per-operator reconciliation incheckParallelismPreconditions.IllegalStateExceptionwhen 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.ChainingMaxParallelismStateLossITCasecovering both directions (explicit value below and above the chain head's) on the HashMap and RocksDB backends.Verifying this change
This change added tests and can be verified as follows:
ChainingMaxParallelismStateLossITCasetakes a canonical savepoint with operator chaining disabled (the keyed operator is its own vertex at an explicit maximum parallelism), then restores with chaining enabled (the keyed operator chains under the source head, whose derived maximum parallelism differs). It asserts the restore is rejected with the new error for both directions and both state backends.StateAssignmentOperationTestandCheckpointCoordinatorRestoringTestcontinue to pass, confirming the new check is additive and runs only on restore.Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Opus 4.8 (1M context)