wenjin272 commented on code in PR #1224:
URL: https://github.com/apache/flink-agents/pull/1224#discussion_r4439500884
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -1636,6 +1651,13 @@ private void tryResumeProcessActionTasks() throws
Exception {
for (int i = 0; i < queued; i++) {
executionCoordinator.addTask(key);
}
+ // Mirror the fresh-record path, which emits one record
start per key: the
+ // worker-completion path emits one record finished per
key, so the resumed
+ // round needs the matching start. A key whose restored
tasks are all synthetic
+ // noop stays silent, because its completion is silent too.
+ if (keysWithVisibleTask.contains(key)) {
+ notifyRecordStart(key);
Review Comment:
With `AGENT_RUN_BEGIN_EVENT` enabled, an Agent that handles
`AgentRunBeginEvent` but has no `InputEvent` handler can restore a queue of
`[beginHandler, internalNoop]`. This condition emits `onRecordStart`, but
`maybeFinishCurrentInput()` suppresses `onRecordFinished` because the last task
is the noop. I reproduced this through checkpoint/restore on JDK 17: the
listener receives only `start:1`.
Could the start and finish paths use the same record-level criterion, with a
regression test for this mixed queue?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -1601,10 +1605,21 @@ private void tryResumeProcessActionTasks() throws
Exception {
// its root caller still awaits it (the caller resumes
only after the child
// quiesces), so the root is always retained here to drive
the replay.
state.update(rootTasks);
+ boolean lifecycleVisible = false;
for (ActionTask actionTask : rootTasks) {
+ // The synthetic noop input action stays out of action
metrics, as it does
+ // on the dequeue path: counting it here would leave
the pending-task gauge
+ // nonzero once the restored task completes.
+ if (isInternalNoopInputAction(actionTask.action)) {
+ continue;
Review Comment:
This also skips the input-run identification performed by
`restoreActionTask()`, as @sangkyoonnam noted. The restored noop carries a
non-null run ID, but completion cannot match it to the anonymous restored slot.
In an operator-level reproduction, restoring a noop and then completing another
fresh input leaves `numOfActiveInputRuns` at 1.
Could we retain input-run identification here while skipping only the
action-level metrics, and cover this with a restore regression test?
--
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]
