GreatEugenius opened a new issue, #1010: URL: https://github.com/apache/flink-agents/issues/1010
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description When the Kafka or Fluss Durable ActionState backend is enabled, restoring a job from a checkpoint can leave duplicated recovery state permanently retained in each subtask's in-memory cache. During recovery: 1. Recovery markers are stored in `UnionListState`, so every restored subtask receives all markers. 2. `KafkaActionStateStore.rebuildState()` merges the markers and scans every referenced Kafka partition. 3. `FlussActionStateStore.rebuildState()` similarly merges the markers and scans every referenced Fluss bucket. 4. As a result, every subtask's `actionStates` map contains the complete recovery tail, including states for keys owned by other subtasks. For example, with parallelism 2: ```text After restore: subtask 0 cache: {A, B} subtask 1 cache: {A, B} Key ownership: subtask 0 owns A subtask 1 owns B ``` When a subsequent checkpoint completes, snapshotLastCompletedSequenceNumbers() scans the current subtask's keyed state only. Therefore, notifyCheckpointComplete() invokes: ```text subtask 0: pruneState(A, ...) subtask 1: pruneState(B, ...) ``` The resulting caches are: ```text subtask 0 cache: {B} subtask 1 cache: {A} ``` The entries belonging to other subtasks are never passed to pruneState() and remain reachable until the current operator attempt is closed or fails again. ### How to reproduce 1. Enable the Kafka or Fluss Durable ActionState backend. 2. Set operator parallelism to 2. 3. Choose keys A and B such that they are assigned to different subtasks. 4. Complete checkpoint C0. 5. After C0, write ActionState records for both A and B. 6. Trigger failover before the next checkpoint completes. 7. Restore from C0. 8. Inspect the `actionStates` map in both subtasks. Both maps contain A and B: ```text subtask 0: {A, B} subtask 1: {A, B} ``` 9. Allow the replayed actions for A and B to complete. 10. Complete checkpoint C1. 11. Inspect both caches again. The current implementation leaves: ```text subtask 0: {B} subtask 1: {A} ``` ### Version and environment Apache Flink Agents 0.2+ ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- 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]
