[Bug] Recovered Durable ActionState for keys owned by other subtasks is never pruned

[da-daken]

Linked issue: #1010

Purpose of 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

• 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

no API

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Cursor 2.4.22 (Claude Opus 4.8), WorkBuddy 5.3.11 (hy3)

[GreatEugenius]

Hi @da-daken, thank you for the PR. I left one comment, which I believe is important.

[GreatEugenius]

The ownership check hashes the String parsed from the durable state key, but Flink assigned keyed-state ownership using the original typed key. KeyGroupRangeAssignment hashes different object types differently; for example, with max parallelism 128, Long(1) maps to key-group 86 while String("1") maps to 54. The true owner can therefore discard its recovered ActionState while another subtask retains it, allowing replay to execute an already completed action again. Please persist the key-group computed from the original typed key in the WAL record and compare that value with currentSubtaskKeyGroupRange, instead of reconstructing ownership from the string form.

[da-daken]

Good catch! Since the string key recovered during rebuild carries no type
information, we can't reconstruct the correct key-group from it (e.g. both
Long(1) and String("1") toString to "1", but hash to different key-groups).
The fix is to store the key-group directly at write time, when the typed key is
still available, so recovery can read it back without relying on the string form.

There are two options for where to store the key-group:

Dimension	A (embed in key)	C (store in ActionState value)
Correctness	Correct. Key-group computed once from typed key at write time.	Correct.
Format change	Key format (4 → 5 segments)	Value format (new field)
Upgrade behavior	Old 4-segment keys are deterministically dropped + warn log. At most one re-execution within the retention window.	Old values without keyGroup field must be dropped, or fall back to the buggy string-hash — which preserves the original bug.
Upgrade cost	One-time: old keys evicted after retention window.	Same one-time cost.
Self-describing	Yes. The key alone tells you which key-group it belongs to.	No. Must read the value to determine ownership.
Identity vs. payload	Key-group is identity information (like business key, seqNum), belongs in the key.	Key-group is stored as payload, mixing identity metadata with business data.
Perf (Fluss rebuild)	Filter by key before deserializing value.	Must deserialize all values to read keyGroup.

Both A and C have the same upgrade cost — old records lack the key-group in
either location and must be dropped. The difference is semantic: key-group is
derived from the typed key at write time, just like the business key and
sequence number. It belongs in the key, not in the value payload.

I prefer A because it keeps identity information in the key where it
belongs, avoids mixing identity metadata with business payload, and in Fluss
avoids deserializing values for non-owned keys during rebuild.

[GreatEugenius]

Thanks for the detailed analysis. Persisting the key-group at write time makes sense.

I am concerned about dropping all legacy 4-segment records during upgrade. A missing key-group does not have to mean either dropping the record or falling back to the incorrect string hash. We could treat it as UNKNOWN ownership:

Records with a key-group: filter normally using the current subtask’s KeyGroupRange.

Legacy records without a key-group: temporarily retain them in every subtask.

This preserves the old memory amplification only for the legacy recovery tail during the first upgraded attempt, while avoiding the loss of durable state. Once a new checkpoint marker advances past those records, they will no longer be included in the next recovery tail.

Dropping legacy records may re-execute every affected action or durable call in the recovery tail, potentially repeating external side effects. For durable execution, I think the bounded one-time memory overhead is preferable to weakening recovery correctness.

[da-daken]

I think it can be kept. To avoid re‑running, the subsequent GET path will also need to be compatible with the 4‑segment key. I'm not sure if adding a fallback logic in the code is acceptable.