sangkyoonnam opened a new pull request, #1245: URL: https://github.com/apache/flink-agents/pull/1245
Linked issue: #1238 ### Purpose of change #### User-visible outcome After a restore, `numOfActiveInputRuns` counts each run that was in flight at the snapshot once. For a single restored in-flight run, the gauge now reads 1 after `open()` and 0 after the run completes; before this change it read 2 and then stayed at 1. #### Runtime flow `tryResumeProcessActionTasks` calls `BuiltInMetrics.restoreActionTask` for each retained task, which adds the task's run ID to `activeInputRunIds`, and then calls `restoreActiveInputRuns(count)` with the number of restored processing keys. `restoreActiveInputRuns` now stores `max(0, count - activeInputRunIds.size())` as unidentified runs, so a run identified from its task is not counted again. `finish(...)` consumes an anonymous slot only for a null run ID, so the extra slot the old code added was never released. #### Key decisions - Fix it in `restoreActiveInputRuns` rather than reorder the operator's restore. The subtraction gives the same gauge whichever call runs first, and the existing count-first test keeps passing. - Clamp at zero so an unexpected surplus of identified runs can't make the anonymous count negative. ### Behavioral Semantics #### Interaction decisions With `n` distinct restored processing keys owned by the subtask: | Restore order | Gauge after restore | After one identified run completes | |---|---:|---:| | Identify runs, then count (the operator's order) | `n` | `n - 1` | | Count, then identify (existing test order) | `n` | `n - 1` | #### Behavioral contracts 1. A restored processing key whose run was identified from a retained task contributes one active input run, not two. 2. Completing an identified restored run decrements the gauge once; when it is the only active run, the gauge reaches zero. #### Failure behavior Unchanged. No new exception, retry or state-format change; only the restored gauge value is corrected. ### Tests | Contract | Coverage | |---|---| | 1. A restored, identified run is counted once | `TaskLifecycleListenerNotificationTest.resumedInFlightRecordCountsAsOneActiveInputRun` (serial engine: snapshot an in-flight record, restore, assert 1 after `open()`); `BuiltInInputRunMetricsTest.restoredRunIdentifiedBeforeCountIsNotCountedTwice` (identify, then count); existing `restoredRunRebuildsActiveGaugeWithoutRecordingHistoricalSamples` (count, then identify) | | 2. Completion decrements once | The same two new tests assert 0 after the run completes; the existing test covers the count-first order | Both new tests fail on main with `expected: 1L but was: 2L` and pass with this change. `mvn -pl runtime test` on JDK 21 passed (1287 run, 0 failures, 28 skipped), and `tools/lint.sh -c` passed on JDK 11. Not verified by restore-gauge assertions: worker-pool and Python task restores. The operator regression exercises serial Java execution; the worker-pool path holds the execution lock through restoration and goes through the same `restoreActiveInputRuns`. ### API No public API changes; only the restored `numOfActiveInputRuns` value is corrected. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Claude Code 2.1.296 (Claude Opus 5.5) -- 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]
