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 6d0a261d5 [runtime][java][python] Keep durable gather outcomes
completed before an ordinary cancellation (#1239)
6d0a261d5 is described below
commit 6d0a261d50c11d3385d520dc2ba635409d0afde0
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Tue Oct 13 15:06:10 2026 +0900
[runtime][java][python] Keep durable gather outcomes completed before an
ordinary cancellation (#1239)
Generated-by: Claude Code 2.1.296 (Claude Opus 5.5)
---
.../flink_agents/runtime/flink_runner_context.py | 44 +++++++++++++++---
.../runtime/tests/test_durable_exception.py | 35 ++++++++++++---
.../runtime/context/JavaRunnerContextImpl.java | 39 ++++++++++------
...vaRunnerContextImplDurableExecuteAsyncTest.java | 52 ++++++++++++++++++++++
4 files changed, 146 insertions(+), 24 deletions(-)
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index 2d68b7906..23e0654c8 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -407,11 +407,38 @@ class _BatchAsyncExecutionResult:
executed = _collect_sliding_window_outcomes_on_timeout(
batch_futures, exception
)
+ # An ordinary interruption must not be cached as a business failure
either.
+ # Durable children before it in input order are still finalized, as in
a
+ # durable-only batch, so recovery does not run a call whose outcome is
known.
+ interrupted = next(
+ (
+ position
+ for position, (index, outcome) in enumerate(
+ zip(execution_indexes, executed, strict=True)
+ )
+ if isinstance(self._futures[index], _SingleAsyncFuture)
+ and is_java_interruption(outcome.error)
+ ),
+ None,
+ )
+ if interrupted is not None:
+ if plan is not None:
+ prefix = [
+ position
+ for position in durable_execution_indexes
+ if position < interrupted
+ ]
+ # The call index stays at the batch base, so re-awaiting
rereads
+ # the saved prefix.
+ self._ctx._finalize_batch_execution(
+ durable_calls,
+ plan,
+ [started[position] for position in prefix],
+ [executed[position] for position in prefix],
+ advance_call_index=False,
+ )
+ raise executed[interrupted].error
for index, outcome in zip(execution_indexes, executed, strict=True):
- if isinstance(
- self._futures[index], _SingleAsyncFuture
- ) and is_java_interruption(outcome.error):
- raise outcome.error
outcomes[index] = outcome
if plan is not None:
durable_outcomes = self._ctx._finalize_batch_execution(
@@ -1490,11 +1517,15 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
plan: _BatchExecutionPlan,
started: list[bool],
executed: list[Outcome],
+ *,
+ advance_call_index: bool = True,
) -> list[Outcome]:
+ # ``executed`` may be a prefix of ``plan.suppliers`` when an ordinary
+ # child in the same batch was interrupted.
base = self._j_runner_context.getCurrentCallIndex()
outcomes = list(plan.outcomes)
for i, ((call_index, _), outcome) in enumerate(
- zip(plan.suppliers, executed, strict=True)
+ zip(plan.suppliers[: len(executed)], executed, strict=True)
):
call = calls[call_index]
function_id = self._durable_identity(call)
@@ -1520,7 +1551,8 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
except Exception as e:
outcome = Outcome.failure(e)
outcomes[call_index] = outcome
- self._j_runner_context.advanceCallIndexBy(len(calls))
+ if advance_call_index:
+ self._j_runner_context.advanceCallIndexBy(len(calls))
return outcomes
@override
diff --git a/python/flink_agents/runtime/tests/test_durable_exception.py
b/python/flink_agents/runtime/tests/test_durable_exception.py
index 8950af89b..3888acdda 100644
--- a/python/flink_agents/runtime/tests/test_durable_exception.py
+++ b/python/flink_agents/runtime/tests/test_durable_exception.py
@@ -677,6 +677,18 @@ def
test_ordinary_java_interruption_propagates_without_caching(
raise interruption
return "retried"
+ durable_calls = []
+
+ def durable() -> str:
+ durable_calls.append(1)
+ return "durable"
+
+ later_calls = []
+
+ def later() -> str:
+ later_calls.append(1)
+ return "later"
+
store = _FakeJavaRunnerContext()
ctx = _create_runner_context(store)
try:
@@ -685,18 +697,25 @@ def
test_ordinary_java_interruption_propagates_without_caching(
handle = future
else:
sibling = (
- ctx.durable_execute_async(lambda: "durable")
+ ctx.durable_execute_async(durable)
if mode == "mixed_batch"
else ctx.execute_async(lambda: "ordinary")
)
- handle = ctx.gather(sibling, future)
+ if mode == "mixed_batch":
+ handle = ctx.gather(sibling, future,
ctx.durable_execute_async(later))
+ else:
+ handle = ctx.gather(sibling, future)
with pytest.raises(type(interruption)) as raised:
_run_async(handle)
assert raised.value is interruption
assert store.current_call_index == 0
+ # The durable sibling completed before the interruption, so its
outcome is
+ # kept and recovery rereads it instead of running it again.
assert [slot.status for slot in store.call_results] == (
- ["PENDING"] if mode == "mixed_batch" else []
+ ["SUCCEEDED", "PENDING"] if mode == "mixed_batch" else []
)
+ if mode == "mixed_batch":
+ assert later_calls == [1]
result = _run_async(handle)
assert (result if mode == "single" else result[1].value) == "retried"
@@ -704,8 +723,14 @@ def
test_ordinary_java_interruption_propagates_without_caching(
assert len(calls) == 2
if mode == "mixed_batch":
assert result[0].value == "durable"
- assert [slot.status for slot in store.call_results] ==
["SUCCEEDED"]
- assert store.current_call_index == 1
+ assert durable_calls == [1]
+ assert result[2].value == "later"
+ assert later_calls == [1, 1]
+ assert [slot.status for slot in store.call_results] == [
+ "SUCCEEDED",
+ "SUCCEEDED",
+ ]
+ assert store.current_call_index == 2
else:
assert store.operations == []
finally:
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
index d4f541637..655368230 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
@@ -144,26 +144,39 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
}
BatchExecutionResult<T> executed = executeOutcomeSuppliers(suppliers);
+ // An ordinary cancellation must not be cached as a business failure
either. Durable
+ // children before it in input order are still finalized, as in a
durable-only batch, so
+ // recovery does not run a call whose outcome is already known.
+ int ordinaryCancellation = executed.getOutcomes().size();
for (int i = 0; i < executed.getOutcomes().size(); i++) {
- Outcome<T> outcome = executed.getOutcomes().get(i);
- // An ordinary cancellation must not be cached as a business
failure either.
- if (futures.get(executableIndexes.get(i)).getDurableCallable() ==
null) {
- rethrowCancellation(outcome.getError());
+ if (futures.get(executableIndexes.get(i)).getDurableCallable() ==
null
+ &&
CancellationUtils.isCancellation(executed.getOutcomes().get(i).getError())) {
+ ordinaryCancellation = i;
+ break;
}
- outcomes.set(executableIndexes.get(i), outcome);
}
List<Outcome<T>> durableOutcomes = new ArrayList<>();
- boolean[] durableStarted = new boolean[durableExecutionIndexes.size()];
- for (int i = 0; i < durableExecutionIndexes.size(); i++) {
- int executionIndex = durableExecutionIndexes.get(i);
+ List<Boolean> durableStarted = new ArrayList<>();
+ for (int executionIndex : durableExecutionIndexes) {
+ if (executionIndex >= ordinaryCancellation) {
+ break;
+ }
durableOutcomes.add(executed.getOutcomes().get(executionIndex));
- durableStarted[i] = executed.wasStarted(executionIndex);
+ durableStarted.add(executed.wasStarted(executionIndex));
+ }
+ boolean[] started = new boolean[durableStarted.size()];
+ for (int i = 0; i < started.length; i++) {
+ started[i] = durableStarted.get(i);
}
finalizeExecutedOutcomes(
- durableCalls,
- base,
- plan,
- new BatchExecutionResult<>(durableOutcomes, durableStarted));
+ durableCalls, base, plan, new
BatchExecutionResult<>(durableOutcomes, started));
+ if (ordinaryCancellation < executed.getOutcomes().size()) {
+ // The call index stays at the batch base, so re-awaiting rereads
the saved prefix.
+
rethrowCancellation(executed.getOutcomes().get(ordinaryCancellation).getError());
+ }
+ for (int i = 0; i < executed.getOutcomes().size(); i++) {
+ outcomes.set(executableIndexes.get(i),
executed.getOutcomes().get(i));
+ }
durableIndex = 0;
for (int i = 0; i < futures.size(); i++) {
if (futures.get(i).getDurableCallable() != null) {
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
index 67ce7c7ec..b1ccc2a44 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
@@ -816,6 +816,58 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
+ " the pending slots instead of skipping them");
}
+ @Test
+ void testMixedGatherOrdinaryInterruptionKeepsEarlierDurableOutcomes()
throws Exception {
+ InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null),
executor);
+ AtomicInteger chargeCalls = new AtomicInteger();
+ AtomicInteger ordinaryCalls = new AtomicInteger();
+ InterruptedException interruption = new
InterruptedException("cancelled");
+ TestDurableCallable<String> charge =
+ new TestDurableCallable<>(
+ "charge",
+ String.class,
+ () -> {
+ chargeCalls.incrementAndGet();
+ return "charged";
+ });
+ TestDurableCallable<String> later =
+ new TestDurableCallable<>("later", String.class, () ->
"later");
+ AsyncFuture<String> ordinary =
+ context.executeAsync(
+ () -> {
+ if (ordinaryCalls.incrementAndGet() == 1) {
+ throw interruption;
+ }
+ return "retried";
+ });
+ AsyncFuture<List<Outcome<String>>> gathered =
+ context.gather(
+ List.of(
+ context.durableExecuteAsync(charge),
+ ordinary,
+ context.durableExecuteAsync(later)));
+ Thread.interrupted(); // clear any stray interrupt flag left over from
another test
+
+ assertSame(interruption, assertThrows(InterruptedException.class,
gathered::await));
+
+ assertTrue(Thread.interrupted());
+ List<CallResult> persisted =
+
context.getDurableExecutionContext().getActionState().getCallResults();
+ assertTrue(persisted.get(0).isSuccess());
+ assertTrue(persisted.get(1).isPending());
+ assertEquals(0,
context.getDurableExecutionContext().getCurrentCallIndex());
+
+ List<Outcome<String>> outcomes = gathered.await();
+
+ assertEquals("charged", outcomes.get(0).getValue());
+ assertEquals("retried", outcomes.get(1).getValue());
+ assertEquals("later", outcomes.get(2).getValue());
+ assertEquals(1, chargeCalls.get());
+ assertEquals(2, ordinaryCalls.get());
+ assertEquals(2,
context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
@Test
void testGatherWithoutDurableStateInterruptionPropagates() throws
Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
