sangkyoonnam opened a new issue, #1238: URL: https://github.com/apache/flink-agents/issues/1238
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description Since #1138, `numOfActiveInputRuns` double-counts restored in-flight input runs identified from retained root tasks. For one such run the gauge reads 2 after restore and stays at 1 after the run completes; it should read 1 and then 0. `tryResumeProcessActionTasks` now calls `builtInMetrics.restoreActionTask(...)` for each retained root task first, which adds each distinct non-null run ID to `activeInputRunIds` through `identifyRestoredActiveInputRun`. Only then does it count the owned processing keys and call `restoreActiveInputRuns(count)`, which sets `unidentifiedRestoredActiveRuns = count` and the gauge to `count + activeInputRunIds.size()`. A run that was just identified is therefore counted a second time as an anonymous restored run. On completion, `finish(...)` removes the identified entry but only consumes an anonymous slot when the ID is null, so the extra slot never goes away. #1138 (`bd527d7f`) moved task restoration ahead of the count. `BuiltInInputRunMetricsTest.restoredRunRebuildsActiveGaugeWithoutRecordingHistoricalSamples` initializes the count before identifying the run, so it doesn't exercise the operator's current restore order. The restore logic is shared by the serial and worker-pool engines. I'd fix this in `restoreActiveInputRuns`, keeping the operator's current order and treating already identified runs as part of the restored processing-key count. For consistent restored state, that works with either order: ```java unidentifiedRestoredActiveRuns = Math.max(0L, count - activeInputRunIds.size()); activeInputRuns.set(unidentifiedRestoredActiveRuns + activeInputRunIds.size()); ``` ### How to reproduce Start from `TaskLifecycleListenerNotificationTest.resumedInFlightRecordReEmitsRecordStart` at 9af407ed, keeping its serial configuration (`PARALLEL_EXECUTION_ENABLED=false`), `SyncAgent` and keyed harness. Read `numOfActiveInputRuns` with the `metricGroup(...)` and `gauge(...)` helpers from `ActionExecutionOperatorParallelMetricsTest`. 1. Open the first harness, process `new StreamRecord<>(7L)`, read the gauge, then call `snapshot(1L, 1L)` before the mailbox runs the queued task. 2. Initialize a new harness from that snapshot, call `open()`, and read the gauge. 3. Call `waitInFlightEventsFinished()` on the restored operator and read the gauge again. ``` before snapshot: numOfActiveInputRuns = 1 after restore/open(): numOfActiveInputRuns = 2 after completion: numOfActiveInputRuns = 1 ``` With the change above the same sequence reads 1, 1, 0, and `BuiltInInputRunMetricsTest`, `TaskLifecycleListenerNotificationTest`, `ActionExecutionOperatorTest` and `ActionExecutionOperatorParallelMetricsTest` still pass. ### Version and environment main at 9af407ed, JDK 17. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
