You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by we...@apache.org on 2026/07/08 07:08:15 UTC
(flink-agents) branch main updated: [hotfix] Propagate action task Errors from mailbox (#880)
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 58801617 [hotfix] Propagate action task Errors from mailbox (#880)
58801617 is described below
commit 588016170be176dab269721bfcd64bbe4c6351e3
Author: Wenjin Xie <16...@users.noreply.github.com>
AuthorDate: Wed Jul 8 15:08:10 2026 +0800
[hotfix] Propagate action task Errors from mailbox (#880)
Catch Throwable in mailbox-submitted action task processing so Errors are rethrown through the mailbox and fail the task visibly instead of remaining in a Future while processing keys stay in-flight.
Co-authored-by: Codex <co...@openai.com>
---
.../runtime/operator/ActionExecutionOperator.java | 6 ++-
.../operator/ActionExecutionOperatorTest.java | 58 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 2 deletions(-)
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
index 665551bb..46d09e0a 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
@@ -283,12 +283,14 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
private void tryProcessActionTaskForKey(Object key) {
try {
processActionTaskForKey(key);
- } catch (Exception e) {
+ } catch (Throwable t) {
+ // MailboxExecutor.submit() stores task failures in its Future. Catch Throwable and
+ // rethrow via execute() so Errors fail the task instead of leaving the key in-flight.
mailboxExecutor.execute(
() ->
ExceptionUtils.rethrow(
new ActionTaskExecutionException(
- "Failed to execute action task", e)),
+ "Failed to execute action task", t)),
"throw exception in mailbox");
}
}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
index 9ac18e3e..22c0072f 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
@@ -289,6 +289,27 @@ public class ActionExecutionOperatorTest {
}
}
+ @Test
+ void testMailboxSubmittedActionTaskPropagatesError() throws Exception {
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory(
+ TestAgent.getLinkageErrorAgentPlan(), true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>) testHarness.getOperator();
+
+ testHarness.processElement(new StreamRecord<>(0L));
+ assertThatThrownBy(() -> operator.waitInFlightEventsFinished())
+ .hasCauseInstanceOf(ActionExecutionOperator.ActionTaskExecutionException.class)
+ .rootCause()
+ .isInstanceOf(NoClassDefFoundError.class)
+ .hasMessageContaining("synthetic missing runtime dependency");
+ }
+ }
+
@Test
void testInMemoryActionStateStoreIntegration() throws Exception {
AgentPlan agentPlanWithStateStore = TestAgent.getAgentPlan(false);
@@ -1711,6 +1732,20 @@ public class ActionExecutionOperatorTest {
}
}
+ public static class LinkageErrorAction {
+ static {
+ if (shouldThrowLinkageError()) {
+ throw new NoClassDefFoundError("synthetic missing runtime dependency");
+ }
+ }
+
+ private static boolean shouldThrowLinkageError() {
+ return true;
+ }
+
+ public static void action(Event event, RunnerContext context) {}
+ }
+
public static void resetReconcilableRecoveryFixture() {
RECONCILABLE_CALL_COUNTER.set(0);
RECONCILABLE_RECONCILE_COUNTER.set(0);
@@ -1937,6 +1972,29 @@ public class ActionExecutionOperatorTest {
return null;
}
+ public static AgentPlan getLinkageErrorAgentPlan() {
+ try {
+ Map<String, List<Action>> actionsByEvent = new HashMap<>();
+ Map<String, Action> actions = new HashMap<>();
+
+ Action errorAction =
+ new Action(
+ "linkageErrorAction",
+ new JavaFunction(
+ LinkageErrorAction.class,
+ "action",
+ new Class<?>[] {Event.class, RunnerContext.class}),
+ Collections.singletonList(InputEvent.EVENT_TYPE));
+ actionsByEvent.put(InputEvent.EVENT_TYPE, Collections.singletonList(errorAction));
+ actions.put(errorAction.getName(), errorAction);
+
+ return new AgentPlan(actions, actionsByEvent, new HashMap<>());
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ return null;
+ }
+
// ==================== Actions for Exception Recovery Tests ====================
/**