dingguotao opened a new pull request, #29893: URL: https://github.com/apache/flink/pull/29893
## 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. - Add `UnalignedCheckpointChainedExchangeITCase` that reproduces the data loss end to end. 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. - `UnalignedCheckpointChainedExchangeITCase`: runs `source(3) -> map(2) -> sink(2)` with unaligned checkpoints and a slow map, waits for a checkpoint with in-flight buffers and restores it as `source(2) -> map(2) -> sink(2)`, where the rebalance exchange becomes a chained forward connection. The restore must either be rejected or every record must reach the sink. Without the fix it fails with e.g. `2947 of 50000 records were lost after recovering from the unaligned checkpoint`; with the fix the restore is rejected. - `StateAssignmentOperationTest` passes locally. This is the `release-1.20` backport of https://github.com/apache/flink/pull/29892. ## 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]
