sangkyoonnam commented on code in PR #1224:
URL: https://github.com/apache/flink-agents/pull/1224#discussion_r4247085239
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -1847,6 +1869,10 @@ public boolean isDone() {
@Override
public void commit() throws Exception {
if (failure != null) {
+ // This engine propagates an action failure through commit()
rather than through
+ // processActionTaskForKey(), so the input run is closed here:
the failure counter,
+ // the active-run gauge and the terminal latency samples all
depend on it.
+
builtInMetrics.markInputRunFailed(actionTask.getTraceContext().getInputRunId());
Review Comment:
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.
```suggestion
if (actionTask != null) {
builtInMetrics.markInputRunFailed(actionTask.getTraceContext().getInputRunId());
}
```
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorParallelMetricsTest.java:
##########
@@ -232,4 +235,36 @@ private static AgentPlan getAgentPlan() throws Exception {
return new AgentPlan(actions, new LinkedHashMap<>(),
parallelConfig());
}
}
+
+ @Test
+ @Timeout(20)
+ void failedParallelActionMarksInputRunFailed() 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);
+
+ assertThatThrownBy(
+ () -> {
+ testHarness.processElement(new
StreamRecord<>(0L));
+ operator.waitInFlightEventsFinished();
+ })
+ .hasStackTraceContaining("parallel action boom");
+
+
assertThat(metricGroup(operator).getCounter("numOfInputRunsFailed").getCount())
+ .as("a failure propagated through commit() must close the
input run")
+ .isEqualTo(1L);
+ }
+ }
Review Comment:
Regression test: fails on 1da2fa63 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.
```suggestion
}
@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);
}
}
```
--
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]
