joeyutong opened a new pull request, #826: URL: https://github.com/apache/flink-agents/pull/826
Linked issue: #824 ### Purpose of change This is the release-0.2 follow-up for #825. This PR fixes duplicate processing-key recovery after rescale or restore. `currentProcessingKeysOpState` is union list state, so every restored subtask sees the union of all previous subtasks' processing keys. The existing restore logic skips non-owned keys when scheduling recovery work, but non-owner subtasks still kept those keys in their local operator state. If a checkpoint is taken before the owner finishes the key, non-owner subtasks can snapshot stale processing keys again. A later restore can then see duplicate entries for the same key, causing duplicate recovery mailbox submissions and eventually making `removeProcessingKey` return a count greater than one. The fix keeps only distinct keys owned by the current subtask in the local operator state after restore. This prevents non-owner subtasks from re-publishing stale keys into later checkpoints. ### Tests - `mvn -pl runtime -am -Dtest=ActionExecutionOperatorTest#testRestoreOnlyResumesKeysOwnedByCurrentSubtask -Dsurefire.failIfNoSpecifiedTests=false test` - `mvn -pl runtime -am -Dtest=ActionExecutionOperatorTest -Dsurefire.failIfNoSpecifiedTests=false test` ### API No public API changes. The change only adds a package-private test helper on the release branch because the release-0.2 operator does not yet have `OperatorStateManager`. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` -- 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]
