dingguotao opened a new pull request, #29892: URL: https://github.com/apache/flink/pull/29892
## What is the purpose of the change Fixes [FLINK-41456](https://issues.apache.org/jira/browse/FLINK-41456). A job restored from an unaligned checkpoint with a different parallelism silently loses the in-flight data of a non-pointwise exchange (e.g. rebalance) when that exchange becomes a chained forward connection in the restored graph. `StateAssignmentOperation#checkForUnsupportedToplogyChanges` already rejects this case, but in `reDistributeResultSubpartitionStates` and `reDistributeInputChannelStates` it runs after the FLINK-31963 short-circuit. The short-circuit only looks at the output (tail) and input (head) operator of the vertex. When the in-flight data is owned by an operator that is chained in the new graph, both conditions are false, the method returns early and the check is skipped, so the persisted records are dropped without an error. ## Brief change log - Run `checkForUnsupportedToplogyChanges` before the FLINK-31963 short-circuit in `reDistributeResultSubpartitionStates` and `reDistributeInputChannelStates`. The short-circuit condition itself is unchanged. The check only iterates over the checkpoint metadata on the JobManager and does not read state files. It only counts handles with offsets as data, so an exchange that becomes chained without in-flight data is still restored as before. ## Verifying this change This change added tests and can be verified as follows: - `StateAssignmentOperationTest#testChainedOperatorWithInFlightOutputDataIsRejected` and `#testChainedOperatorWithInFlightInputDataIsRejected`: a chained vertex whose non-tail / non-head operator carries output / input channel state with offsets is rejected. Both fail without the fix. - `StateAssignmentOperationTest#testChainedOperatorWithoutInFlightDataIsAccepted`: the same topology with empty channel state handles is still accepted. - `StateAssignmentOperationTest`, `TaskStateAssignmentTest` and `CheckpointCoordinatorRestoringTest` pass locally. The end-to-end reproduction (an ITCase) is described in the Jira issue. It is written against `release-1.20` and is part of the backport PR; an ITCase for master will follow separately, since the 1.20 one uses the legacy `SinkFunction` API. ## 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 (restore from unaligned checkpoints that previously dropped in-flight data now fails explicitly) - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Opus 5.5 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
