[CodeTrainerMan]

Linked issue: #1218

Purpose of change

User-visible outcome

On the JDK < 21 parallel action execution engine (no coroutines):

1. Listeners registered on the operator now observe a paired onRecordStart /
   onRecordFinished bracket after a failover, instead of a onRecordFinished with no
   matching start.
2. Restoring a checkpoint that contains a queued synthetic no-op task no longer leaves
   numOfPendingActionTasks permanently nonzero after the task completes.
3. An action failure now closes its input run: numOfInputRunsFailed is incremented, the
   active-run gauge is decremented, and terminal samples for inputRunLatencyMs and
   inputRunProcessingLatencyMs are recorded.
4. The listener contracts no longer promise mailbox-thread callbacks, which the parallel
   engine does not honor.

Runtime flow

On restore, tryResumeProcessActionTasks walks the keyed action-task state, keeps only root
tasks, restores their metrics and, per owned key, requeues one coordinator node per queued
task. The fix adds a record-start notification in that per-key branch, so the bracket opened
here is closed by maybeFinishCurrentInput, which emits one onRecordFinished per key when
the last in-flight task of the key commits. The same loop now skips the synthetic no-op input
action when restoring metrics, matching what the dequeue path already does.

On the failure path, the parallel engine runs an action to completion off the mailbox thread
and propagates its failure from Work.commit(). That is the only place the failure is
observed on this engine, so the input run is closed there.

Key decisions

• Emit onRecordStart once per key, not once per requeued task. Completion emits one
  onRecordFinished per key, so a per-task start would open more brackets than are closed.
• Gate the notification on the key having at least one lifecycle-visible restored task. A key
  whose restored tasks are all synthetic no-op completes silently, so starting it would open
  a bracket that is never closed.
• Reuse the existing isInternalNoopInputAction predicate for the restore path rather than
  introducing a second notion of "lifecycle-visible", so the dequeue and restore paths cannot
  disagree.
• Close the failed input run inside commit() rather than at the call site, because
  commit() is where this engine observes the failure; markInputRunFailed is idempotent,
  so it cannot double count a run that was already closed.

Behavioral Semantics

Interaction decisions

Engine	Restored task	Record start on restore	Record finished on completion	Pending-task gauge
Parallel (JDK < 21)	at least one non-noop	emitted once per key	emitted once per key	returns to zero
Parallel (JDK < 21)	only synthetic no-op	not emitted	not emitted	unaffected
Serial	any	emitted once per key (unchanged)	emitted once per key	unchanged

Failure path:

Engine	Action failure	numOfInputRunsFailed	Exception propagated
Parallel (JDK < 21)	raised by the action	incremented once	ActionTaskExecutionException, unchanged
Serial	raised by the action	incremented once (unchanged)	unchanged

Behavioral contracts

1. After a failover on the parallel engine, a lifecycle listener observes onRecordStart for
   a resumed key before the onRecordFinished that completion emits for that key.
2. A key whose restored tasks are all synthetic no-op emits neither record start nor record
   finished.
3. After a checkpoint containing a queued synthetic no-op task is restored and that task
   completes, numOfPendingActionTasks is zero.
4. When a parallel action fails, numOfInputRunsFailed increases by one for that input run,
   and the active-input-run gauge is decremented once.
5. The documented threading contract of TaskLifecycleListener and
   ComponentExecutionListener states that the callback thread is not fixed, so a listener
   that keeps state must synchronize it itself.

Failure behavior

• Action failure on the parallel engine: the input run is closed and the failure is still
  rethrown as ActionTaskExecutionException; nothing is swallowed or retried.
• A failure whose input run is unknown or already closed is absorbed by
  BuiltInInputRunMetrics.finish, which removes the run id before recording, so no gauge is
  decremented twice and no latency sample is recorded twice.
• No change to the listener exception contract: a throwing listener still fails the task on
  the serial path and is still logged and swallowed for component execution reporting.

Tests

Contracts to tests

Contract	Test
1. paired start/finished after failover	none — see Not verified
2. all-noop key stays silent	none — see Not verified
3. pending-task gauge returns to zero	none — see Not verified
4. failed input run is counted	ActionExecutionOperatorParallelMetricsTest.failedParallelActionMarksInputRunFailed
5. documented threading contract	none — the change is prose

Coverage by risk

The change touches restore-time bookkeeping and the failure path. Contract 4 is pinned by a
new component test that runs only when continuations are unsupported, i.e. exactly the engine
the change targets. The existing parallel and lifecycle suites
(ActionExecutionOperatorTest, ParallelExecutionCoordinatorTest,
ParallelExecutionLockTest, ParallelExecutionTaskQueueTest,
TaskLifecycleListenerNotificationTest, ActionExecutionOperatorCapacityIntegrationTest,
ActionExecutionOperatorDrainDeadlockIntegrationTest) cover the surrounding behavior and were
run with the change.

Not verified

• Contracts 1–3 have no automated test. The harness cannot hold a queued task across a
  snapshot on this engine: the parallel engine drains in-flight tasks before checkpointing,
  so a snapshot taken while a worker is blocked waits for that worker, and a snapshot taken
  after it is released leaves no queued task behind. The serial restore path is covered by
  TaskLifecycleListenerNotificationTest.resumedInFlightRecordReEmitsRecordStart; the
  parallel restore path is not.
• No coverage for a failure raised after the input run was already closed, and none for a
  restore whose queued tasks mix synthetic no-op and ordinary tasks on one key.

Implementation invariants and supporting evidence

• markInputRunFailed is idempotent: BuiltInInputRunMetrics.finish removes the run id from
  activeInputRunIds and activeRunTimings before recording, and returns early when the
  timing is absent, so a repeated call neither decrements the gauge twice nor records a
  second sample.
• Work.commit() is invoked only from ParallelExecutionCoordinator, so the serial catch
  block that already calls markInputRunFailed cannot run for the same failure.
• The synthetic no-op task is excluded from onRecordFinished by the completion path and from
  markActionTaskDequeued by the dequeue path; the restore path now matches both.
• A key with no restored action-task state is absent from the visible-task set and therefore
  emits no start, which matches it never emitting a finish.

API

Compatibility impact

No public API changes. TaskLifecycleListener and ComponentExecutionListener keep their
signatures and default methods; only their contract documentation changes, and it now
describes behavior the parallel engine already had. What does change for an existing caller
is observable metrics: failed input runs on the parallel engine are now counted and released,
and the pending-task gauge no longer retains restored synthetic no-op tasks. Callers that
relied on the previous counts observe different values; listeners that assumed mailbox-thread
affinity were already incorrect on this engine.

Documentation

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

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

• Yes
• No

[sangkyoonnam]

The new markInputRunFailed call in Work.commit() turns a preparation failure into an NPE that hides the original exception.

Work.actionTask is assigned only after preparation: runner context creation, onActionPrepared listeners, durable state setup and the continuation context lookup. When any of those throws, execute() records the failure with actionTask still null, and commit() now dereferences it. On JDK 17, with a listener whose onActionPrepared throws, the test in the inline suggestion fails on 1da2fa6 with Cannot invoke "org.apache.flink.agents.runtime.operator.ActionTask.getTraceContext()" because "this.actionTask" is null. On bd527d7 the same failure surfaces as ActionTaskExecutionException caused by the listener's exception.

The fix has two parts. One is outside the diff, so it can't be a suggestion: in Work.execute(), record the task as soon as it is polled.

                checkState(task != null, "Action task queue was empty.");
                this.actionTask = task;

The later assignment then becomes redundant. The commit() guard and a regression test are suggestions inline. With all three, the test passes and numOfInputRunsFailed is 1. That matches the serial path for preparation exceptions: it reads the input run id right after polling and marks the run failed if preparation throws an Exception. ActionExecutionOperatorTest, ActionExecutionOperatorParallelMetricsTest, TaskLifecycleListenerNotificationTest and ParallelExecutionCoordinatorTest pass with the change (98 tests), and tools/lint.sh -c is clean on JDK 11.

For contract 1 (restored record start/finish pairing), the harness can retain a queued task on the parallel engine. ActionExecutionOperatorTest.restoreRebuildsInFlightBudgetFromProcessingKeysAndPendingEvents snapshots with key 2 queued behind the busy worker, because pre-barrier draining pauses new pulls and waits for already-pulled tasks to commit. Restoring that snapshot through a factory that registers a recording listener before open(), like ListenerInjectingOperatorFactory in TaskLifecycleListenerNotificationTest, looks like it would cover contract 1. I haven't tried it. Would you rather add that here or in a follow-up?

[sangkyoonnam]

With the execute() change from the review body, this guard is still needed for failures before actionTask is assigned, including pollNextActionTask() throwing or the empty-queue check failing.

Suggested change

	builtInMetrics.markInputRunFailed(actionTask.getTraceContext().getInputRunId());
	if (actionTask != null) {
	builtInMetrics.markInputRunFailed(actionTask.getTraceContext().getInputRunId());
	}

[sangkyoonnam]

Regression test: fails on 1da2fa6 with the NPE, passes with the execute() change and the guard above. The listener type is fully qualified so this applies without an import change outside the diff.

Suggested change

	}
	}
	@Test
	@Timeout(20)
	void preparationFailureKeepsCauseAndMarksInputRunFailed() throws Exception {
	assumeThat(ContinuationActionExecutor.isContinuationSupported()).isFalse();
	try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness =
	newHarness(new AgentPlan(new FailingAgent(), parallelConfig()))) {
	testHarness.open();
	ActionExecutionOperator<Long, Object> operator = operator(testHarness);
	operator.addTaskLifecycleListener(
	new org.apache.flink.agents.runtime.lifecycle.TaskLifecycleListener() {
	@Override
	public void onActionPrepared(ActionTask task) {
	throw new IllegalStateException("prepare boom");
	}
	});
	assertThatThrownBy(
	() -> {
	testHarness.processElement(new StreamRecord<>(0L));
	operator.waitInFlightEventsFinished();
	})
	.hasStackTraceContaining("prepare boom");
	assertThat(metricGroup(operator).getCounter("numOfInputRunsFailed").getCount())
	.isEqualTo(1L);
	}
	}

[sangkyoonnam]

One more restore interaction, related to #1238. In the restore path, restoreActionTask identifies run ids before restoreActiveInputRuns sets the anonymous restored-key count, and the fix proposed in #1238 subtracts the already identified runs from that count.

At 1da2fa6 the new noop continue also skips identifying the input run. For a restore with one processing key, no pending inputs and one retained noop task with a non-null run id, numOfActiveInputRuns reads 1 after restore and stays 1 after completion, because finish(...) only consumes an anonymous slot for a null id. On main at 9af407e the same case reads 2 then 1 because of the double count in #1238. With the proposed #1238 fix, main would read 1 then 0, while this PR would still read 1 then 1.

Could the noop path still identify the run and only skip the action-level metrics, for example through a run-only method on BuiltInMetrics? I haven't run this restore case; the values follow from the code at 1da2fa6 and 9af407e.

[CodeTrainerMan]

Thanks @sangkyoonnam — both suggestions are applied and the regression test now matches your sketch.

1. execute(): this.actionTask = task; is now bound right after pollNextActionTask() (and the empty-queue checkState), so it is non-null on a preparation failure.
2. commit() guard: kept exactly as you suggested —
   if (actionTask != null) { builtInMetrics.markInputRunFailed(actionTask.getTraceContext().getInputRunId()); }
   It is still required for the failures that happen before assignment (e.g. pollNextActionTask() throwing or the empty-queue check failing). With the early binding, a preparation-phase failure now also carries a non-null task and still marks the input run failed, so the counter is correct.

Regression test: preparationFailureKeepsCauseAndMarksInputRunFailed uses operator.addTaskLifecycleListener(...) with onActionPrepared throwing. It asserts the original cause is preserved (hasStackTraceContaining("prepare boom")) and numOfInputRunsFailed == 1L. It fails on 1da2fa63 with the exact NPE you quoted and passes now (5/5 in ActionExecutionOperatorParallelMetricsTest).

The earlier CI "Java format check failed" was only import ordering in the new test (a non-static import sat below the static imports); fixed via spotless:apply, and spotless:check is green.

[sangkyoonnam]

Thanks, the preparation-failure fix looks good to me. At 4089c0c preparationFailureKeepsCauseAndMarksInputRunFailed and the rest of ActionExecutionOperatorParallelMetricsTest pass locally on JDK 17 (5/5; the class skips itself on JDK 21, where continuations are supported).

The restore case from my later comment is still open. Take the parallel engine with one owned processing key, no pending inputs and a single retained noop task. The noop continue skips restoreActionTask, leaving the task's run unidentified. On completion, maybeFinishCurrentInput passes the task's non-null run id to markInputRunCompleted, which can't consume the anonymous restored slot. I checked this with a metrics-level replay of that call order: restoreActionTask for the noop task or not, then restoreActiveInputRuns(1), then markInputRunCompleted with the task's run id. With #1245's counting change and the identification call, numOfActiveInputRuns reads 1 after restore and 0 after completion. Without the call, as in this PR, it stays at 1 after completion even with #1245.

Could BuiltInMetrics expose the existing BuiltInInputRunMetrics.identifyRestoredActiveInputRun, with the restore loop calling it for the noop task before the continue and still skipping its action-level metrics? If you'd rather leave it to #1245, I can handle it there.

[wenjin272]

Thanks @CodeTrainerMan for working on this, and sorry for the late review. The preparation-failure fix looks good. I left two inline comments on the remaining restore cases; could you take a look?

[wenjin272]

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?

[wenjin272]

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?