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

[GreatEugenius]

Search before asking

• I searched in the 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:

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:

subtask 0: pruneState(A, ...)
subtask 1: pruneState(B, ...)

The resulting caches are:

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:

   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:

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!