This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new a0f8351c [runtime] Count restored input runs once when tasks identify
them first (#1245)
a0f8351c is described below
commit a0f8351c4b6164392e0254169aef7025ca304025
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Fri Oct 16 13:05:54 2026 +0900
[runtime] Count restored input runs once when tasks identify them first
(#1245)
Generated-by: Claude Code 2.1.296 (Claude Opus 5.5)
---
.../runtime/metrics/BuiltInInputRunMetrics.java | 4 +-
.../TaskLifecycleListenerNotificationTest.java | 47 ++++++++++++++++++++++
.../metrics/BuiltInInputRunMetricsTest.java | 17 ++++++++
3 files changed, 67 insertions(+), 1 deletion(-)
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetrics.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetrics.java
index 390bf3bc..9c4480e7 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetrics.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetrics.java
@@ -129,7 +129,9 @@ final class BuiltInInputRunMetrics {
}
void restoreActiveInputRuns(long count) {
- unidentifiedRestoredActiveRuns = Math.max(0L, count);
+ // The operator identifies runs from restored action tasks before it
knows the count, so
+ // the already identified runs are part of count rather than in
addition to it.
+ unidentifiedRestoredActiveRuns = Math.max(0L, count -
activeInputRunIds.size());
activeInputRuns.set(unidentifiedRestoredActiveRuns +
activeInputRunIds.size());
}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/lifecycle/TaskLifecycleListenerNotificationTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/lifecycle/TaskLifecycleListenerNotificationTest.java
index d2baee41..1693af27 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/lifecycle/TaskLifecycleListenerNotificationTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/lifecycle/TaskLifecycleListenerNotificationTest.java
@@ -32,6 +32,7 @@ import
org.apache.flink.agents.runtime.actionstate.ActionState;
import org.apache.flink.agents.runtime.actionstate.ActionStateStore;
import org.apache.flink.agents.runtime.actionstate.InMemoryActionStateStore;
import org.apache.flink.agents.runtime.async.ContinuationActionExecutor;
+import org.apache.flink.agents.runtime.metrics.FlinkAgentsMetricGroupImpl;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperator;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
import org.apache.flink.agents.runtime.operator.ActionTask;
@@ -45,6 +46,7 @@ import
org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -415,6 +417,51 @@ public class TaskLifecycleListenerNotificationTest {
}
}
+ @Test
+ void resumedInFlightRecordCountsAsOneActiveInputRun() throws Exception {
+ AgentConfiguration serialConfig = new AgentConfiguration();
+ serialConfig.set(AgentExecutionOptions.PARALLEL_EXECUTION_ENABLED,
false);
+ AgentPlan plan = new AgentPlan(new SyncAgent(), serialConfig);
+ OperatorSubtaskState snapshot;
+
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(plan, true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ testHarness.processElement(new StreamRecord<>(7L));
+ snapshot = testHarness.snapshot(1L, 1L);
+ }
+
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(plan, true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.initializeState(snapshot);
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>)
testHarness.getOperator();
+ Field metricGroupField =
ActionExecutionOperator.class.getDeclaredField("metricGroup");
+ metricGroupField.setAccessible(true);
+ FlinkAgentsMetricGroupImpl metricGroup =
+ (FlinkAgentsMetricGroupImpl)
metricGroupField.get(operator);
+
+ assertThat(
+ ((Number)
metricGroup.getGauge("numOfActiveInputRuns").getValue())
+ .longValue())
+ .isEqualTo(1L);
+
+ operator.waitInFlightEventsFinished();
+
+ assertThat(
+ ((Number)
metricGroup.getGauge("numOfActiveInputRuns").getValue())
+ .longValue())
+ .isZero();
+ }
+ }
+
@Test
void recordWithoutTriggeredActionsEmitsNoLifecycleEvents() throws
Exception {
AgentPlan plan = new AgentPlan(new EmptyAgent());
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetricsTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetricsTest.java
index fb38c894..3d0b306c 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetricsTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/metrics/BuiltInInputRunMetricsTest.java
@@ -103,6 +103,23 @@ class BuiltInInputRunMetricsTest {
assertThat(gauge(BuiltInInputRunMetrics.NUM_ACTIVE_INPUT_RUNS)).isZero();
}
+ @Test
+ void restoredRunIdentifiedBeforeCountIsNotCountedTwice() {
+ setTimeMillis(300L);
+ ExecutionTraceContext inputRun =
ExecutionTraceContext.forInputRun("key", "agent");
+
+ // The operator restores action tasks, identifying their runs, before
it counts the
+ // restored processing keys.
+ metrics.identifyRestoredActiveInputRun(inputRun.getInputRunId());
+ metrics.restoreActiveInputRuns(1L);
+
assertThat(gauge(BuiltInInputRunMetrics.NUM_ACTIVE_INPUT_RUNS)).isEqualTo(1L);
+
+ setTimeMillis(375L);
+ metrics.inputRunCompleted(inputRun.getInputRunId());
+
+
assertThat(gauge(BuiltInInputRunMetrics.NUM_ACTIVE_INPUT_RUNS)).isZero();
+ }
+
@Test
void
restoredPendingInputRecordsLocallyObservedOutcomeAndProcessingLatency() {
Event inputEvent = new InputEvent("restored-pending");
