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 6323cbd1 [runtime] Replay a durable call that returned null instead of
throwing (#1171)
6323cbd1 is described below
commit 6323cbd196a5dc6d27efe6a301be79c4514c20d1
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Tue Sep 29 16:27:16 2026 +0900
[runtime] Replay a durable call that returned null instead of throwing
(#1171)
Generated-by: Claude Code 2.1.283 (Claude Opus 5.5)
---
.../agents/runtime/context/RunnerContextImpl.java | 21 ++++++++----
...vaRunnerContextImplDurableExecuteAsyncTest.java | 21 ++++++++++++
.../RunnerContextImplDurableExecuteTest.java | 40 ++++++++++++++++++++++
3 files changed, 75 insertions(+), 7 deletions(-)
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
index ec0eccec..e09d50b8 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
@@ -620,10 +620,10 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
return executeAndFinalizeCurrentCall(functionId, argsDigest,
executionCallable);
}
- Optional<T> cachedResult =
+ Optional<Outcome<T>> cachedResult =
tryGetCachedResult(functionId, argsDigest,
durableCallable.getResultClass());
if (cachedResult.isPresent()) {
- return cachedResult.get();
+ return cachedResult.get().getValue();
}
T result = null;
@@ -916,7 +916,13 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
return null;
}
- protected <T> Optional<T> tryGetCachedResult(
+ /**
+ * Returns the recorded outcome of the next durable call when it matches
and has finished, or an
+ * empty {@code Optional} when there is nothing to replay. A recorded
failure is rethrown. A
+ * recorded success may carry a {@code null} value (for example a {@code
Void} call), so the
+ * presence of the outcome, not its value, marks the hit.
+ */
+ protected <T> Optional<Outcome<T>> tryGetCachedResult(
String functionId, String argsDigest, Class<T> resultClass) throws
Exception {
Object[] cached = matchNextOrClearSubsequentCallResult(functionId,
argsDigest);
if (cached != null && (Boolean) cached[0]) {
@@ -928,9 +934,10 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
OBJECT_MAPPER.readValue(exceptionPayload,
DurableExecutionException.class);
throw cachedException.toException();
} else if (resultPayload != null) {
- return Optional.of(OBJECT_MAPPER.readValue(resultPayload,
resultClass));
+ return Optional.of(
+ Outcome.success(OBJECT_MAPPER.readValue(resultPayload,
resultClass)));
} else {
- return Optional.of(null);
+ return Optional.of(Outcome.success(null));
}
}
return Optional.empty();
@@ -978,10 +985,10 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
}
if (!current.isPending()) {
- Optional<T> cachedResult =
+ Optional<Outcome<T>> cachedResult =
tryGetCachedResult(functionId, argsDigest,
durableCallable.getResultClass());
if (cachedResult.isPresent()) {
- return cachedResult.get();
+ return cachedResult.get().getValue();
}
throw new IllegalStateException(
String.format(
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 401374d0..35f9ed08 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
@@ -146,6 +146,27 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
assertEquals(0, persistCallCount.get());
}
+ @Test
+ void testDurableExecuteAsyncReconcilableReplayNullSuccess() throws
Exception {
+ InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(new CallResult("recon-async", "", null,
null));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestReconcilableCallable<Void> callable =
+ new TestReconcilableCallable<>(
+ "recon-async",
+ Void.class,
+ () -> fail("call should not be executed"),
+ () -> fail("reconcile should not be called for
terminal slot"));
+
+ assertNull(context.durableExecuteAsync(callable).await());
+ assertEquals(0, callable.getCallCount());
+ assertEquals(0, callable.getReconcileCount());
+ assertEquals(0, executor.getExecuteAsyncCallCount());
+ assertEquals(0, persistCallCount.get());
+ assertEquals(1,
context.getDurableExecutionContext().getActionState().getCallResultCount());
+ }
+
@Test
void testDurableExecuteAsyncReconcilableReconcileSuccess() throws
Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
index b85ea929..2cad205e 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextImplDurableExecuteTest.java
@@ -105,6 +105,26 @@ class RunnerContextImplDurableExecuteTest {
assertEquals(1, persistCallCount.get());
}
+ @Test
+ void nullResultReplaysThroughRuntimeStateWithoutReExecution() throws
Exception {
+ RunnerContextImpl context = createContext(new ActionState(new
Event("test")));
+ TestDurableCallable<Void> callable =
+ new TestDurableCallable<>("void-call", Void.class, () -> null);
+ assertNull(context.durableExecute(callable));
+
+ ActionState restored =
+
org.apache.flink.agents.runtime.actionstate.ActionStateSerde.deserialize(
+
org.apache.flink.agents.runtime.actionstate.ActionStateSerde.serialize(
+ lastPersistedState));
+ RunnerContextImpl replay = createContext(restored);
+
+ assertNull(replay.durableExecute(callable));
+ assertEquals(1, callable.getCallCount());
+ assertEquals(1, persistCallCount.get());
+ assertEquals(1,
replay.getDurableExecutionContext().getCurrentCallIndex());
+ assertEquals(1,
replay.getDurableExecutionContext().getActionState().getCallResultCount());
+ }
+
@Test
void testDurableExecuteCompletionOnlyDoesNotPersistInterruption() {
RunnerContextImpl context = createContext(new ActionState(null));
@@ -288,6 +308,26 @@ class RunnerContextImplDurableExecuteTest {
assertEquals(1,
context.getDurableExecutionContext().getCurrentCallIndex());
}
+ @Test
+ void testDurableExecuteReconcilableReplayNullSuccess() throws Exception {
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(new CallResult("recon-call", "", null,
null));
+ RunnerContextImpl context = createContext(actionState);
+ TestReconcilableCallable<String> callable =
+ new TestReconcilableCallable<>(
+ "recon-call",
+ String.class,
+ () -> fail("call should not be re-executed"),
+ () -> fail("reconcile should not be called for
terminal slot"));
+
+ assertNull(context.durableExecute(callable));
+ assertEquals(0, callable.getCallCount());
+ assertEquals(0, callable.getReconcileCount());
+ assertEquals(0, persistCallCount.get());
+ assertEquals(1,
context.getDurableExecutionContext().getCurrentCallIndex());
+ assertEquals(1,
context.getDurableExecutionContext().getActionState().getCallResultCount());
+ }
+
@Test
void testDurableExecuteReconcilableReplayFailure() throws Exception {
ActionState actionState = new ActionState(null);
