This is an automated email from the ASF dual-hosted git repository. wenjin272 pushed a commit to branch release-0.3 in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 09b81387d62e73a89ebcb4ea9a28362d5ee49315 Author: Wenjin Xie <[email protected]> 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 <[email protected]> (cherry picked from commit 588016170be176dab269721bfcd64bbe4c6351e3) --- .../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 ==================== /**
