GreatEugenius commented on code in PR #1024:
URL: https://github.com/apache/flink-agents/pull/1024#discussion_r3817881126
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -578,10 +579,22 @@ public void initializeState(StateInitializationContext
context) throws Exception
super.initializeState(context);
durableExecManager.maybeInitActionStateStore(agentPlan.getConfig());
- durableExecManager.handleRecovery(getOperatorStateBackend());
stateManager = new OperatorStateManager();
+ // Drop action-state records owned by other subtasks during rebuild.
UnionListState
+ // broadcasts every subtask's recovery marker, so a naive replay would
load all keys into
+ // every subtask's cache, where the foreign ones are never pruned
(orphan-state leak).
+ int maxParallelism =
getRuntimeContext().getTaskInfo().getMaxNumberOfParallelSubtasks();
+ KeyGroupRange currentSubtaskKeyGroupRange =
+ stateManager.getCurrentSubtaskKeyGroupRange(maxParallelism,
getRuntimeContext());
+ Predicate<String> ownershipFilter =
+ key ->
+ stateManager.isKeyOwnedByCurrentSubtask(
Review Comment:
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.
--
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]
