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 24c992353 [runtime][python] Propagate durable call result persistence
failures (#1211)
24c992353 is described below
commit 24c992353e1fe59520db7f8c3ac9b89616915256
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Fri Oct 9 12:53:52 2026 +0900
[runtime][python] Propagate durable call result persistence failures (#1211)
Generated-by: Claude Code 2.1.292 (Claude Opus 5.5)
---
.../flink_agents/runtime/flink_runner_context.py | 15 ++--
.../test_flink_runner_context_reconcilable.py | 87 ++++++++++++++++++++++
2 files changed, 95 insertions(+), 7 deletions(-)
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index 048ba04bc..140d5b29c 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -969,14 +969,15 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
result_payload, exception_payload = self._serialize_call_payloads(
result, exception
)
-
- self._j_runner_context.recordCallCompletion(
- function_id, result_payload, exception_payload
- )
except Exception as e:
- # If Java method doesn't exist, silently ignore
- if "recordCallCompletion" not in str(e):
- logger.warning("Failed to record call completion: %s", e)
+ logger.warning("Failed to record call completion: %s", e)
+ return
+
+ # A failed write must reach the caller: returning normally would let
the
+ # action proceed although recovery could execute the call again.
+ self._j_runner_context.recordCallCompletion(
+ function_id, result_payload, exception_payload
+ )
@staticmethod
def _serialize_call_payloads(
diff --git
a/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
b/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
index 7c272e5eb..308639a26 100644
---
a/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
+++
b/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
@@ -1353,3 +1353,90 @@ def
test_flink_runner_context_gather_respects_max_parallelism() -> None:
"SUCCEEDED",
]
assert j_runner_context.current_call_index == 4
+
+
+class _FailingPersistJavaRunnerContext(_FakeJavaRunnerContext):
+ """Fails every call-result write the way ``DurableExecutionManager``
does."""
+
+ def recordCallCompletion(
+ self,
+ function_id: str,
+ result_payload: bytes | None,
+ exception_payload: bytes | None,
+ ) -> None:
+ self.operations.append("record")
+ msg = "java.lang.RuntimeException: Failed to persist ActionState"
+ raise RuntimeError(msg)
+
+
+def _durable_call(ctx: FlinkRunnerContext, asynchronous: bool, func: Any) ->
Any:
+ if asynchronous:
+ return _run_async(ctx.durable_execute_async(func))
+ return ctx.durable_execute(func)
+
+
[email protected]("asynchronous", [False, True])
+def test_call_result_persistence_failure_reaches_the_caller(
+ asynchronous: bool,
+) -> None:
+ calls = []
+
+ def call() -> str:
+ calls.append(1)
+ return "receipt-1"
+
+ store = _FailingPersistJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(RuntimeError, match="Failed to persist
ActionState"):
+ _durable_call(ctx, asynchronous, call)
+ finally:
+ _close_runner_context(ctx)
+
+ assert len(calls) == 1
+ assert store.operations.count("record") == 1
+ assert store.call_results == []
+
+
[email protected]("asynchronous", [False, True])
+def test_result_that_cannot_be_pickled_is_returned_unrecorded(
+ asynchronous: bool,
+) -> None:
+ lock = threading.Lock()
+
+ def call() -> Any:
+ return lock
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ result = _durable_call(ctx, asynchronous, call)
+ finally:
+ _close_runner_context(ctx)
+
+ assert result is lock
+ assert "record" not in store.operations
+ assert store.call_results == []
+
+
[email protected]("asynchronous", [False, True])
+def test_persistence_failure_replaces_the_call_exception(asynchronous: bool)
-> None:
+ calls = []
+
+ def call() -> str:
+ calls.append(1)
+ msg = "call failed"
+ raise ValueError(msg)
+
+ store = _FailingPersistJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(RuntimeError, match="Failed to persist
ActionState") as info:
+ _durable_call(ctx, asynchronous, call)
+ finally:
+ _close_runner_context(ctx)
+
+ assert info.value.__cause__ is None
+ assert len(calls) == 1
+ assert store.operations.count("record") == 1
+ assert store.call_results == []
