da-daken opened a new pull request, #1024: URL: https://github.com/apache/flink-agents/pull/1024
<!-- * Thank you very much for contributing to Flink Agents. * Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc. --> <!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. --> Linked issue: https://github.com/apache/flink-agents/issues/1010 ### Purpose of change <!-- What is the purpose of this change? --> I've confirmed the root cause: `rebuildState` replays all recovery markers (UnionListState broadcasts every subtask's marker) into each subtask's `actionStates` cache, while `notifyCheckpointComplete` only prunes keys present in the current subtask's own keyed state — so keys owned by other subtasks are never pruned and stay resident for the whole operator attempt. **Adopted fix**: keep the bulk replay, but skip records not owned by the current subtask during rebuild, so foreign keys never enter the cache. #### Changes - **`ActionStateStore`**: add default method `setOwnershipFilter(Predicate<String>)` (`null` means no filter, safe for in-memory/test backends). - **`KafkaActionStateStore` / `FlussActionStateStore`**: hold the predicate; in `rebuildState`'s replay loop, skip records whose business key is not owned by the current subtask, reusing the existing `OperatorStateManager.isKeyOwnedByCurrentSubtask` (Flink key-group semantics). - **`DurableExecutionManager.handleRecovery`**: compute the current subtask's `KeyGroupRange` and set it as the filter before `rebuildState`. - **`ActionExecutionOperator.initializeState`**: compute `maxParallelism` + `KeyGroupRange` (already available here) and pass them into `handleRecovery`. ### Tests <!-- How is this change verified? --> - **Unit — `ActionStateUtilTest`**: backend-agnostic coverage of `isKeyRetained` (owned key retained, foreign key dropped, `null` filter keeps all, unparseable key retained). Both stores route through this function, so this is the shared filtering contract. - **Unit — `KafkaActionStateStoreTest`** (MockConsumer): seed records for keys A/B, set the filter to "owner of A only", assert the cache keeps only A; plus no-filter-keeps-all and unparseable-key-retained cases. - **Integration — `FlussActionStateStoreIT.testRebuildStateFiltersForeignKeys`** (real Fluss cluster via `FlussClusterExtension`): write A and B, capture a recovery marker, then rebuild into a fresh store with the filter "owner of A only" and assert A is recovered while B is filtered out. This is the only test that drives Fluss's real log-scan → deserialize → filter → cache replay path. ### API <!-- Does this change touches any public APIs? --> no API ### Documentation <!-- Do not remove this section. Check the proper box only. --> - [ ] `doc-needed` <!-- Your PR changes impact docs --> - [x] `doc-not-needed` <!-- Your PR changes do not impact docs --> - [ ] `doc-included` <!-- Your PR already contains the necessary documentation updates --> ### Was this patch authored or co-authored using generative AI tooling? <!-- Do not remove this section. Check the proper box only. --> - [x] Yes - [ ] No Generated-by: Cursor 2.4.22 (Claude Opus 4.8) -- 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]
