CodeTrainerMan opened a new pull request, #1224:
URL: https://github.com/apache/flink-agents/pull/1224
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.
<details>
<summary>Implementation invariants and supporting evidence</summary>
- `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.
</details>
### 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`
- [x] `doc-not-needed`
- [ ] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [ ] Yes
- [x] No
--
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]
