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 66f23e478 [runtime] Process the input queued behind an input that
triggers no Actions (#1310)
66f23e478 is described below
commit 66f23e478d0726c4740c5a4f1025fcb7ef376e11
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Thu Oct 22 10:57:30 2026 +0900
[runtime] Process the input queued behind an input that triggers no Actions
(#1310)
Generated-by: Claude Code 2.1.303 (Claude Opus 5.5)
---
.../runtime/operator/ActionExecutionOperator.java | 6 ++++
.../operator/ActionExecutionOperatorTest.java | 40 ++++++++++++++++++++++
2 files changed, 46 insertions(+)
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 68a85ee22..b05f30231 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
@@ -770,6 +770,12 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
"Current key" + key + " is missing from the
segmentedQueue.");
eventRouter.processEligibleWatermarks(super::processWatermark);
builtInMetrics.markInputRunCompleted(currentInputRunId);
+ // An Action condition can leave an input without tasks; start
the input queued
+ // behind it, as a finished input does.
+ Event pendingInputEvent = pollNextPendingInputEvent();
+ if (pendingInputEvent != null) {
+ processInputEvent(key, pendingInputEvent);
+ }
return;
}
currentInputRunId = actionTask.getTraceContext().getInputRunId();
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 fac682483..abe0196a1 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
@@ -630,6 +630,46 @@ public class ActionExecutionOperatorTest {
}
}
+ @Test
+ @Timeout(10)
+ void
nonParallelEnginesProcessInputQueuedBehindInputWithoutTriggeredActions() throws
Exception {
+ AgentConfiguration config = new AgentConfiguration();
+ config.set(AgentExecutionOptions.PARALLEL_EXECUTION_ENABLED, false);
+ Action action =
+ new Action(
+ "pending-events-single",
+ new JavaFunction(
+ TestAgent.class,
+ "pendingEventsSingleAction",
+ new Class<?>[] {Event.class,
RunnerContext.class}),
+ Collections.singletonList("type ==
EventType.InputEvent && input != 2"));
+ Map<String, Action> actions = new HashMap<>();
+ actions.put(action.getName(), action);
+ AgentPlan agentPlan = new AgentPlan(actions, new HashMap<>(), config);
+
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory(agentPlan, true),
+ (KeySelector<Long, Long>) value -> 0L,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>)
testHarness.getOperator();
+
+ // Inputs 2 and 3 queue behind input 1 for the same key; input 2
triggers no Actions.
+ testHarness.processElement(new StreamRecord<>(1L));
+ testHarness.processElement(new StreamRecord<>(2L));
+ testHarness.processElement(new StreamRecord<>(3L));
+ operator.endInput();
+ operator.waitInFlightEventsFinished();
+
+ assertThat(testHarness.getRecordOutput())
+ .extracting(StreamRecord::getValue)
+ .containsExactly("single-1", "single-3");
+
assertThat(operator.getOperatorStateManager().getProcessingKeys()).isEmpty();
+ }
+ }
+
@Test
@Timeout(30)
void backpressureBlocksAdmissionAtCapAndReleasesOnRetirement() throws
Exception {
