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 550d50def [runtime][python] Keep durable call failures whose exception
can't be pickled or rebuilt (#1182)
550d50def is described below
commit 550d50defa76d988a144df697a7059278f5121fe
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Fri Oct 9 11:42:36 2026 +0900
[runtime][python] Keep durable call failures whose exception can't be
pickled or rebuilt (#1182)
Generated-by: Claude Code 2.1.284 (Claude Fable 5.1)
---
docs/content/docs/development/workflow_agent.md | 1 +
python/flink_agents/api/runner_context.py | 8 +
python/flink_agents/runtime/durable_exception.py | 207 +++++++
.../flink_agents/runtime/flink_runner_context.py | 48 +-
.../runtime/tests/test_durable_exception.py | 659 +++++++++++++++++++++
.../test_flink_runner_context_reconcilable.py | 5 +-
6 files changed, 920 insertions(+), 8 deletions(-)
diff --git a/docs/content/docs/development/workflow_agent.md
b/docs/content/docs/development/workflow_agent.md
index 264cec27d..48f1a6243 100644
--- a/docs/content/docs/development/workflow_agent.md
+++ b/docs/content/docs/development/workflow_agent.md
@@ -473,6 +473,7 @@ on how to setup and configure the external action state
store.
- Results may not be reused if call order or arguments change
(non-deterministic actions), which clears subsequent cached results and
re-executes.
- If a failure happens after a function starts but before it completes and its
result is persisted, the call will be re-executed. See the "With a reconciler"
section below.
- In Python async actions, if `ctx.durable_execute_async(...)` is not awaited,
the result is not recorded and cannot be replayed.
+- In Python, a recorded failure is replayed as the original exception when it
can be pickled and rebuilt. Otherwise it is replayed as a `RuntimeError` with
the message `<class>: <message>`, so code that catches a specific exception
type may take a different path during replay. A failure that cannot be pickled
is recorded this way too, so the call is not executed again on recovery.
Failures recorded by a version that stored only the pickled exception have no
class name or message to fall [...]
**With a reconciler:**
diff --git a/python/flink_agents/api/runner_context.py
b/python/flink_agents/api/runner_context.py
index 5b5cdc84c..d14e171a7 100644
--- a/python/flink_agents/api/runner_context.py
+++ b/python/flink_agents/api/runner_context.py
@@ -100,6 +100,8 @@ class DurableFuture(ABC, Generic[T]):
except Exception as error:
# Control-flow BaseExceptions leave the handle unresolved so a
later
# await or gather cannot mistake cancellation for a durable
outcome.
+ if self._is_cancellation(error):
+ raise
self._error = error
self._done = True
raise
@@ -117,6 +119,12 @@ class DurableFuture(ABC, Generic[T]):
self._error = outcome.error
self._done = True
+ def _is_cancellation(self, error: Exception) -> bool:
+ """Return whether ``error`` reports a cancelled attempt rather than the
+ outcome of the call. Such an error leaves the handle unresolved.
+ """
+ return False
+
@abstractmethod
def _resolve(self) -> Any:
"""Resolve this future and return a generator consumed by
``__await__``."""
diff --git a/python/flink_agents/runtime/durable_exception.py
b/python/flink_agents/runtime/durable_exception.py
new file mode 100644
index 000000000..ecdf6d8ae
--- /dev/null
+++ b/python/flink_agents/runtime/durable_exception.py
@@ -0,0 +1,207 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+"""Persisted form of an exception raised by a durable call.
+
+An exception is stored as a marker followed by a pickled dict that keeps the
class
+name and message next to the pickled exception. Pickle alone is not enough:
some
+exceptions cannot be pickled, and others pickle but cannot be rebuilt. With the
+class name and message stored separately, replay can always raise something
that
+names the recorded failure.
+"""
+
+import logging
+from typing import Any
+
+import cloudpickle
+
+logger = logging.getLogger(__name__)
+
+# A pickle stream never starts with these bytes, so a payload without them is
one
+# written before this format existed: the pickled exception itself.
+_MARKER = b"FLINK-AGENTS-EXC\x00"
+_VERSION = 1
+# The message is stored next to the pickle, which usually holds it too. The
copy is
+# bounded so a long message does not double the record; the fallback only needs
+# enough of it to name the failure.
+_MAX_MESSAGE_LENGTH = 16 * 1024
+_TRUNCATION_MARK = "... [truncated]"
+
+# java.io.InterruptedIOException is left out on purpose. HTTP stacks raise it
for an
+# ordinary network timeout, which must stay a recorded failure. Same choice as
+# ModelRoutingResolver.isCancellation on the Java side.
+_JAVA_INTERRUPTION_CLASS_NAMES = frozenset(
+ {
+ "java.lang.InterruptedException",
+ "java.nio.channels.ClosedByInterruptException",
+ }
+)
+_MAX_DEPTH = 64
+
+
+def _class_name(exception: BaseException) -> str:
+ cls = type(exception)
+ return f"{cls.__module__}.{cls.__qualname__}"
+
+
+def durable_exception_message(exception: BaseException) -> str:
+ """Return the message of ``exception``, tolerating a formatter that
raises."""
+ try:
+ return str(exception)
+ except Exception:
+ return "<message unavailable>"
+
+
+def _recorded_message(exception: BaseException) -> str:
+ message = durable_exception_message(exception)
+ if len(message) > _MAX_MESSAGE_LENGTH:
+ return message[:_MAX_MESSAGE_LENGTH] + _TRUNCATION_MARK
+ return message
+
+
+def serialize_durable_exception(exception: BaseException) -> bytes:
+ """Serialize ``exception`` for the durable call record. Never raises for an
+ exception that cannot be pickled; the class name and message are kept
instead.
+ """
+ try:
+ pickled = cloudpickle.dumps(exception)
+ except Exception:
+ logger.debug(
+ "Durable call exception %s cannot be pickled; "
+ "recording its class name and message only.",
+ _class_name(exception),
+ exc_info=True,
+ )
+ pickled = None
+ return _MARKER + cloudpickle.dumps(
+ {
+ "version": _VERSION,
+ "class": _class_name(exception),
+ "message": _recorded_message(exception),
+ "pickle": pickled,
+ }
+ )
+
+
+def deserialize_durable_exception(payload: bytes) -> BaseException:
+ """Rebuild the exception recorded for a durable call.
+
+ Returns the original exception when it can be rebuilt, and otherwise a
+ ``RuntimeError`` carrying the recorded class name and message.
+ """
+ payload = bytes(payload)
+ if not payload.startswith(_MARKER):
+ return cloudpickle.loads(payload)
+
+ try:
+ record = cloudpickle.loads(payload[len(_MARKER) :])
+ version = record["version"]
+ except Exception as e:
+ msg = "The recorded durable call exception cannot be read."
+ raise RuntimeError(msg) from e
+ if type(version) is not int or version != _VERSION:
+ msg = f"Unsupported durable call exception record version: {version}"
+ raise RuntimeError(msg)
+ try:
+ class_name = record["class"]
+ message = record["message"]
+ pickled = record["pickle"]
+ except Exception as e:
+ msg = "The recorded durable call exception is incomplete."
+ raise RuntimeError(msg) from e
+ if not (
+ isinstance(class_name, str)
+ and isinstance(message, str)
+ and (pickled is None or isinstance(pickled, bytes))
+ ):
+ msg = "The recorded durable call exception is malformed."
+ raise RuntimeError(msg)
+
+ if pickled is not None:
+ try:
+ exception = cloudpickle.loads(pickled)
+ except Exception:
+ logger.debug(
+ "Recorded durable call exception %s cannot be rebuilt; "
+ "raising its class name and message instead.",
+ class_name,
+ exc_info=True,
+ )
+ else:
+ if isinstance(exception, BaseException):
+ return exception
+ return RuntimeError(f"{class_name}: {message}")
+
+
+def _is_pemja_object(obj: Any) -> bool:
+ cls = type(obj)
+ return cls.__name__ == "PyJObject" and cls.__module__.split(".")[0] ==
"pemja"
+
+
+def _java_class_is_interruption(j_class: Any) -> bool:
+ depth = 0
+ while j_class is not None and depth < _MAX_DEPTH:
+ if str(j_class.getName()) in _JAVA_INTERRUPTION_CLASS_NAMES:
+ return True
+ j_class = j_class.getSuperclass()
+ depth += 1
+ return False
+
+
+def _java_throwable_is_interruption(obj: Any) -> bool:
+ try:
+ depth = 0
+ while obj is not None and depth < _MAX_DEPTH and _is_pemja_object(obj):
+ if _java_class_is_interruption(obj.getClass()):
+ return True
+ obj = obj.getCause()
+ depth += 1
+ except BaseException:
+ # Not a throwable, or the Java call failed: nothing to conclude from
it.
+ # BaseException is caught on purpose, here and in
is_java_interruption: the
+ # caller is about to record or re-raise a failure, and anything raised
while
+ # looking at it would replace that failure.
+ logger.debug("Failed to inspect a Java object.", exc_info=True)
+ return False
+
+
+def is_java_interruption(exception: BaseException | None) -> bool:
+ """Whether ``exception`` carries a Java throwable that is, or was caused
by, a
+ thread interruption.
+
+ Pemja raises a Java exception in Python as ``RuntimeError`` whose only
argument
+ is the Java throwable, so the throwable's class and cause chain are
checked,
+ never the message. Such a failure belongs to a cancelled attempt and must
not
+ be recorded as the result of the call.
+
+ Only the explicit cause chain (``raise ... from error``) is followed. The
+ implicit context is not: a failure raised while an interruption was being
+ handled is a failure of its own. An exception that cannot be inspected is
not
+ an interruption, and the inspection never replaces it with a failure of its
+ own.
+ """
+ try:
+ current = exception
+ depth = 0
+ while current is not None and depth < _MAX_DEPTH:
+ if any(_java_throwable_is_interruption(arg) for arg in
current.args):
+ return True
+ current = current.__cause__
+ depth += 1
+ except BaseException:
+ logger.debug("Failed to inspect a durable call exception.",
exc_info=True)
+ return False
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index c385b7b82..048ba04bc 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -45,6 +45,12 @@ from flink_agents.api.runner_context import (
RunnerContext,
)
from flink_agents.api.trace import ExecutionReporter
+from flink_agents.runtime.durable_exception import (
+ deserialize_durable_exception,
+ durable_exception_message,
+ is_java_interruption,
+ serialize_durable_exception,
+)
from flink_agents.runtime.durable_execution import (
_validate_reconciler_callable,
durable_identity_for_call,
@@ -175,7 +181,7 @@ class _DurableExecutionException(Exception):
exception: BaseException,
record_callback: Callable,
) -> None:
- super().__init__(str(exception))
+ super().__init__(durable_exception_message(exception))
self.func = func
self.args = args
self.kwargs = kwargs
@@ -382,6 +388,10 @@ class _SingleDurableFuture(DurableFuture[Any]):
self._ctx = ctx
self._call = call
+ @override
+ def _is_cancellation(self, error: Exception) -> bool:
+ return is_java_interruption(error)
+
def _resolve(self) -> Any:
return (yield from
self._ctx._resolve_durable_call(self._call).__await__())
@@ -394,6 +404,10 @@ class _GatherDurableFuture(DurableFuture[list[Outcome]]):
self._ctx = ctx
self._futures = futures
+ @override
+ def _is_cancellation(self, error: Exception) -> bool:
+ return is_java_interruption(error)
+
def _resolve(self) -> Any:
outcomes_by_index: dict[int, Outcome] = {}
unresolved_futures = []
@@ -901,7 +915,9 @@ class FlinkRunnerContext(RunnerContext, ExecutionReporter):
if is_hit:
if exception_payload is not None:
# Store cached exception to re-raise outside try block
- cached_exception =
cloudpickle.loads(bytes(exception_payload))
+ cached_exception = deserialize_durable_exception(
+ exception_payload
+ )
elif result_payload is not None:
return True, cloudpickle.loads(bytes(result_payload))
else:
@@ -942,11 +958,17 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
exception : BaseException | None
The exception raised by the function (None if successful).
"""
+ if is_java_interruption(exception):
+ # The attempt was cancelled: record nothing, so recovery executes
the
+ # call again instead of replaying the interruption.
+ return
+
function_id = durable_identity_for_call(func, args, kwargs)
try:
- result_payload = None if exception else cloudpickle.dumps(result)
- exception_payload = cloudpickle.dumps(exception) if exception else
None
+ result_payload, exception_payload = self._serialize_call_payloads(
+ result, exception
+ )
self._j_runner_context.recordCallCompletion(
function_id, result_payload, exception_payload
@@ -962,7 +984,9 @@ class FlinkRunnerContext(RunnerContext, ExecutionReporter):
exception: BaseException | None,
) -> tuple[bytes | None, bytes | None]:
result_payload = None if exception else cloudpickle.dumps(result)
- exception_payload = cloudpickle.dumps(exception) if exception else None
+ exception_payload = (
+ serialize_durable_exception(exception) if exception else None
+ )
return result_payload, exception_payload
def _read_call_result_at(self, index: int) -> _PersistedCallResult | None:
@@ -1012,6 +1036,11 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
result: Any,
exception: BaseException | None,
) -> None:
+ if is_java_interruption(exception):
+ # The attempt was cancelled: leave the slot pending, so recovery
+ # executes or reconciles the call instead of replaying the
interruption.
+ return
+
function_id = durable_identity_for_call(func, args, kwargs)
result_payload, exception_payload = self._serialize_call_payloads(
result,
@@ -1203,7 +1232,9 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
try:
if current.exception_payload is not None:
- return
Outcome.failure(cloudpickle.loads(current.exception_payload))
+ return Outcome.failure(
+ deserialize_durable_exception(current.exception_payload)
+ )
if current.result_payload is None:
return Outcome.success(None)
return Outcome.success(cloudpickle.loads(current.result_payload))
@@ -1290,6 +1321,11 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
if not started[i]:
outcomes[call_index] = outcome
continue
+ if is_java_interruption(outcome.error):
+ # The attempt was cancelled: leave this slot and the ones
after it
+ # pending and propagate, as the Java batch does. Slots
finalized
+ # earlier in this loop keep their outcome.
+ raise outcome.error
try:
result_payload, exception_payload =
self._serialize_call_payloads(
outcome.value,
diff --git a/python/flink_agents/runtime/tests/test_durable_exception.py
b/python/flink_agents/runtime/tests/test_durable_exception.py
new file mode 100644
index 000000000..1c2e37571
--- /dev/null
+++ b/python/flink_agents/runtime/tests/test_durable_exception.py
@@ -0,0 +1,659 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import threading
+from typing import Any
+
+import cloudpickle
+import pytest
+
+from flink_agents.runtime.durable_exception import (
+ deserialize_durable_exception,
+ is_java_interruption,
+ serialize_durable_exception,
+)
+from flink_agents.runtime.tests.test_flink_runner_context_reconcilable import (
+ _close_runner_context,
+ _create_runner_context,
+ _FakeJavaRunnerContext,
+ _preload_pending,
+ _run_async,
+)
+
+
+class _KeywordOnlyError(Exception):
+ """Pickles, but cannot be rebuilt: unpickling calls ``cls(*args)``."""
+
+ def __init__(self, message: str, *, status: int) -> None:
+ super().__init__(message)
+ self.status = status
+
+
+class _UnpicklableError(Exception):
+ def __init__(self, message: str) -> None:
+ super().__init__(message)
+ self.lock = threading.Lock()
+
+
+_KEYWORD_ONLY_NAME = f"{__name__}._KeywordOnlyError"
+_UNPICKLABLE_NAME = f"{__name__}._UnpicklableError"
+
+
+def _refuse_pickling(self: Any) -> None:
+ msg = "cannot pickle 'pemja.PyJObject' object"
+ raise TypeError(msg)
+
+
+def _fake_java_throwable(class_name: str, cause: Any = None) -> Any:
+ """Build an object shaped like the ``pemja.PyJObject`` of a Java
throwable: it
+ prints as the Java throwable does and cannot be pickled.
+ """
+
+ def j_class(name: str | None) -> Any:
+ if name is None:
+ return None
+ parent = "java.lang.Exception" if name != "java.lang.Exception" else
None
+ return type(
+ "PyJObject",
+ (),
+ {
+ "__module__": "pemja",
+ "getName": lambda self: name,
+ "getSuperclass": lambda self: j_class(parent),
+ },
+ )()
+
+ return type(
+ "PyJObject",
+ (),
+ {
+ "__module__": "pemja",
+ "__reduce_ex__": lambda self, protocol: _refuse_pickling(self),
+ "__str__": lambda self: f"{class_name}: call failed",
+ "getClass": lambda self: j_class(class_name),
+ "getCause": lambda self: cause,
+ },
+ )()
+
+
+_INTERRUPTED = "java.lang.InterruptedException"
+_CLOSED_BY_INTERRUPT = "java.nio.channels.ClosedByInterruptException"
+_INTERRUPTED_IO = "java.io.InterruptedIOException"
+_ILLEGAL_STATE = "java.lang.IllegalStateException"
+
+
+def _java_error(class_name: str, cause: Any = None) -> RuntimeError:
+ return RuntimeError(_fake_java_throwable(class_name, cause))
+
+
+def _execute(ctx: Any, asynchronous: bool, func: Any, **kwargs: Any) -> Any:
+ if asynchronous:
+ return _run_async(ctx.durable_execute_async(func, **kwargs))
+ return ctx.durable_execute(func, **kwargs)
+
+
+def _run_twice(
+ store: _FakeJavaRunnerContext, asynchronous: bool, func: Any, **kwargs: Any
+) -> list[BaseException]:
+ """Execute the call, then replay it in a new context over the same
store."""
+ raised = []
+ for _ in range(2):
+ store.current_call_index = 0
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(BaseException) as error:
+ _execute(ctx, asynchronous, func, **kwargs)
+ raised.append(error.value)
+ finally:
+ _close_runner_context(ctx)
+ return raised
+
+
+def test_picklable_exception_round_trips_with_its_type() -> None:
+ restored = deserialize_durable_exception(
+ serialize_durable_exception(ValueError("boom"))
+ )
+
+ assert type(restored) is ValueError
+ assert restored.args == ("boom",)
+
+
+def test_exception_that_cannot_be_rebuilt_falls_back_to_class_and_message() ->
None:
+ restored = deserialize_durable_exception(
+ serialize_durable_exception(_KeywordOnlyError("rate limited",
status=429))
+ )
+
+ assert type(restored) is RuntimeError
+ assert str(restored) == f"{_KEYWORD_ONLY_NAME}: rate limited"
+
+
+def test_exception_that_cannot_be_pickled_falls_back_to_class_and_message() ->
None:
+ restored = deserialize_durable_exception(
+ serialize_durable_exception(_UnpicklableError("payment declined"))
+ )
+
+ assert type(restored) is RuntimeError
+ assert str(restored) == f"{_UNPICKLABLE_NAME}: payment declined"
+
+
+def test_payload_written_before_the_envelope_still_loads() -> None:
+ restored =
deserialize_durable_exception(cloudpickle.dumps(ValueError("old")))
+
+ assert type(restored) is ValueError
+ assert restored.args == ("old",)
+
+
[email protected]("asynchronous", [False, True])
+def test_replay_of_exception_that_cannot_be_rebuilt(asynchronous: bool) ->
None:
+ calls = []
+
+ def call() -> None:
+ calls.append(1)
+ msg = "rate limited"
+ raise _KeywordOnlyError(msg, status=429)
+
+ store = _FakeJavaRunnerContext()
+ first, replayed = _run_twice(store, asynchronous, call)
+
+ assert type(first) is _KeywordOnlyError
+ assert type(replayed) is RuntimeError
+ assert str(replayed) == f"{_KEYWORD_ONLY_NAME}: rate limited"
+ assert len(calls) == 1
+ assert [slot.status for slot in store.call_results] == ["FAILED"]
+
+
+def test_gather_replays_exception_that_cannot_be_rebuilt() -> None:
+ calls = []
+
+ def call() -> None:
+ calls.append(1)
+ msg = "rate limited"
+ raise _KeywordOnlyError(msg, status=429)
+
+ store = _FakeJavaRunnerContext()
+ errors = []
+ for _ in range(2):
+ store.current_call_index = 0
+ ctx = _create_runner_context(store)
+ try:
+ outcomes = _run_async(ctx.gather(ctx.durable_execute_async(call)))
+ errors.append(outcomes[0].error)
+ finally:
+ _close_runner_context(ctx)
+
+ assert type(errors[0]) is _KeywordOnlyError
+ assert type(errors[1]) is RuntimeError
+ assert str(errors[1]) == f"{_KEYWORD_ONLY_NAME}: rate limited"
+ assert len(calls) == 1
+
+
[email protected]("asynchronous", [False, True])
+def test_exception_that_cannot_be_pickled_is_recorded(asynchronous: bool) ->
None:
+ calls = []
+
+ def call() -> None:
+ calls.append(1)
+ msg = "payment declined"
+ raise _UnpicklableError(msg)
+
+ store = _FakeJavaRunnerContext()
+ first, replayed = _run_twice(store, asynchronous, call)
+
+ assert type(first) is _UnpicklableError
+ assert type(replayed) is RuntimeError
+ assert str(replayed) == f"{_UNPICKLABLE_NAME}: payment declined"
+ assert len(calls) == 1
+ assert [slot.status for slot in store.call_results] == ["FAILED"]
+
+
[email protected]("asynchronous", [False, True])
[email protected]("use_reconciler", [False, True])
+def test_pending_slot_keeps_exception_that_cannot_be_pickled(
+ use_reconciler: bool, asynchronous: bool
+) -> None:
+ def call() -> None:
+ msg = "payment declined"
+ raise _UnpicklableError(msg)
+
+ def reconciler() -> None:
+ msg = "payment declined"
+ raise _UnpicklableError(msg)
+
+ store = _FakeJavaRunnerContext()
+ _preload_pending(store, call)
+ ctx = _create_runner_context(store)
+ kwargs = {"reconciler": reconciler} if use_reconciler else {}
+ try:
+ with pytest.raises(_UnpicklableError, match="payment declined"):
+ _execute(ctx, asynchronous, call, **kwargs)
+ finally:
+ _close_runner_context(ctx)
+
+ assert [slot.status for slot in store.call_results] == ["FAILED"]
+ restored =
deserialize_durable_exception(store.call_results[0].exception_payload)
+ assert str(restored) == f"{_UNPICKLABLE_NAME}: payment declined"
+
+
[email protected](
+ "error",
+ [
+ _java_error(_INTERRUPTED),
+ _java_error(_CLOSED_BY_INTERRUPT),
+ _java_error("java.lang.RuntimeException",
_fake_java_throwable(_INTERRUPTED)),
+ ],
+ ids=["direct", "closed-by-interrupt", "wrapped"],
+)
+def test_java_interruption_is_recognized(error: RuntimeError) -> None:
+ wrapped = ConnectionError("call failed")
+ wrapped.__cause__ = error
+
+ assert is_java_interruption(error)
+ assert is_java_interruption(wrapped)
+
+
[email protected](
+ "error",
+ [
+ _java_error(_ILLEGAL_STATE),
+ # An ordinary network timeout is an InterruptedIOException too.
+ _java_error(_INTERRUPTED_IO),
+ RuntimeError("java.lang.InterruptedException: sleep interrupted"),
+ ValueError("boom"),
+ None,
+ ],
+ ids=["other-java", "interrupted-io", "message-only", "python", "none"],
+)
+def test_other_failures_are_not_java_interruptions(error: Any) -> None:
+ assert not is_java_interruption(error)
+
+
+def test_java_interruption_is_not_recorded() -> None:
+ calls = []
+
+ def call() -> str:
+ calls.append(1)
+ if len(calls) == 1:
+ raise _java_error(_INTERRUPTED)
+ return "done"
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(RuntimeError):
+ ctx.durable_execute(call)
+ finally:
+ _close_runner_context(ctx)
+ assert store.call_results == []
+
+ store.current_call_index = 0
+ ctx = _create_runner_context(store)
+ try:
+ assert ctx.durable_execute(call) == "done"
+ finally:
+ _close_runner_context(ctx)
+ assert len(calls) == 2
+ assert [slot.status for slot in store.call_results] == ["SUCCEEDED"]
+
+
[email protected]("use_reconciler", [False, True])
+def test_java_interruption_leaves_pending_slot(use_reconciler: bool) -> None:
+ def call() -> None:
+ raise _java_error(_INTERRUPTED)
+
+ store = _FakeJavaRunnerContext()
+ _preload_pending(store, call)
+ ctx = _create_runner_context(store)
+ kwargs = {"reconciler": call} if use_reconciler else {}
+ try:
+ with pytest.raises(RuntimeError):
+ ctx.durable_execute(call, **kwargs)
+ finally:
+ _close_runner_context(ctx)
+
+ assert [slot.status for slot in store.call_results] == ["PENDING"]
+ assert store.current_call_index == 0
+
+
+def test_other_java_failure_is_recorded() -> None:
+ calls = []
+
+ def call() -> None:
+ calls.append(1)
+ raise _java_error(_ILLEGAL_STATE)
+
+ store = _FakeJavaRunnerContext()
+ first, replayed = _run_twice(store, False, call)
+
+ assert type(first) is RuntimeError
+ assert type(replayed) is RuntimeError
+ assert str(replayed) == f"builtins.RuntimeError: {_ILLEGAL_STATE}: call
failed"
+ assert len(calls) == 1
+ assert [slot.status for slot in store.call_results] == ["FAILED"]
+
+
+class _UnprintableError(Exception):
+ def __str__(self) -> str:
+ msg = "formatter failed"
+ raise ValueError(msg)
+
+
+def test_exception_whose_message_cannot_be_formatted_round_trips() -> None:
+ restored = deserialize_durable_exception(
+ serialize_durable_exception(_UnprintableError())
+ )
+
+ assert type(restored) is _UnprintableError
+
+
[email protected]("pending", [False, True])
+def test_exception_whose_message_cannot_be_formatted_is_recorded(
+ pending: bool,
+) -> None:
+ def call() -> None:
+ raise _UnprintableError
+
+ store = _FakeJavaRunnerContext()
+ if pending:
+ _preload_pending(store, call)
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(_UnprintableError):
+ ctx.durable_execute(call)
+ finally:
+ _close_runner_context(ctx)
+
+ assert [slot.status for slot in store.call_results] == ["FAILED"]
+
+
+def test_non_ascii_message_survives_the_fallback() -> None:
+ restored = deserialize_durable_exception(
+ serialize_durable_exception(_UnpicklableError("결제 거절: café"))
+ )
+
+ assert str(restored) == f"{_UNPICKLABLE_NAME}: 결제 거절: café"
+
+
[email protected]("protocol", range(cloudpickle.pickle.HIGHEST_PROTOCOL
+ 1))
+def test_payload_written_with_any_pickle_protocol_is_read_as_legacy(
+ protocol: int,
+) -> None:
+ payload = cloudpickle.dumps(ValueError("old"), protocol=protocol)
+
+ restored = deserialize_durable_exception(payload)
+
+ assert type(restored) is ValueError
+
+
[email protected]("wrapper", [bytes, bytearray, memoryview])
+def test_payload_is_read_from_any_bytes_like_object(wrapper: Any) -> None:
+ payload = wrapper(serialize_durable_exception(ValueError("boom")))
+
+ assert type(deserialize_durable_exception(payload)) is ValueError
+
+
[email protected](
+ "payload",
+ [
+ b"FLINK-AGENTS-EXC\x00not a pickle",
+ b"FLINK-AGENTS-EXC\x00" + cloudpickle.dumps(["not", "a", "record"]),
+ b"FLINK-AGENTS-EXC\x00" + cloudpickle.dumps({"version": 99}),
+ b"FLINK-AGENTS-EXC\x00" + cloudpickle.dumps({"version": 1}),
+ b"FLINK-AGENTS-EXC\x00"
+ + cloudpickle.dumps(
+ {"version": 1, "class": None, "message": None, "pickle": None}
+ ),
+ b"FLINK-AGENTS-EXC\x00"
+ + cloudpickle.dumps(
+ {"version": True, "class": "E", "message": "m", "pickle": None}
+ ),
+ ],
+ ids=[
+ "corrupt",
+ "not-a-record",
+ "unknown-version",
+ "incomplete",
+ "malformed",
+ "boolean-version",
+ ],
+)
+def test_unreadable_record_is_reported(payload: bytes) -> None:
+ with pytest.raises(RuntimeError, match="durable call exception"):
+ deserialize_durable_exception(payload)
+
+
+def test_failure_raised_while_handling_an_interruption_is_not_one() -> None:
+ try:
+ try:
+ raise _java_error(_INTERRUPTED)
+ except RuntimeError:
+ msg = "declined"
+ raise ValueError(msg) # noqa: B904
+ except ValueError as error:
+ failure = error
+
+ assert failure.__context__ is not None
+ assert not is_java_interruption(failure)
+
+
+def test_interruption_is_found_past_an_argument_that_cannot_be_inspected() ->
None:
+ not_a_throwable = type("PyJObject", (), {"__module__": "pemja"})()
+ error = ConnectionError(not_a_throwable)
+ error.__cause__ = _java_error(_INTERRUPTED)
+
+ assert is_java_interruption(error)
+
+
[email protected]("use_reconciler", [False, True])
+def test_java_interruption_is_not_recorded_by_async_call(
+ use_reconciler: bool,
+) -> None:
+ def call() -> None:
+ raise _java_error(_INTERRUPTED)
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ kwargs = {"reconciler": call} if use_reconciler else {}
+ try:
+ with pytest.raises(RuntimeError):
+ _run_async(ctx.durable_execute_async(call, **kwargs))
+ finally:
+ _close_runner_context(ctx)
+
+ expected = ["PENDING"] if use_reconciler else []
+ assert [slot.status for slot in store.call_results] == expected
+ assert store.current_call_index == 0
+
+
[email protected]("asynchronous", [False, True])
[email protected]("use_reconciler", [False, True])
+def test_java_interruption_leaves_preloaded_pending_slot(
+ use_reconciler: bool, asynchronous: bool
+) -> None:
+ interruption = _java_error(_INTERRUPTED)
+
+ def call() -> None:
+ raise interruption
+
+ store = _FakeJavaRunnerContext()
+ _preload_pending(store, call)
+ ctx = _create_runner_context(store)
+ kwargs = {"reconciler": call} if use_reconciler else {}
+ try:
+ with pytest.raises(RuntimeError) as raised:
+ _execute(ctx, asynchronous, call, **kwargs)
+ finally:
+ _close_runner_context(ctx)
+
+ assert raised.value is interruption
+ assert [slot.status for slot in store.call_results] == ["PENDING"]
+ assert store.current_call_index == 0
+
+
+def test_interrupted_future_is_executed_again_when_awaited_again() -> None:
+ calls = []
+
+ def call() -> str:
+ calls.append(1)
+ if len(calls) == 1:
+ raise _java_error(_INTERRUPTED)
+ return "done"
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ future = ctx.durable_execute_async(call)
+ with pytest.raises(RuntimeError):
+ _run_async(future)
+ outcomes = _run_async(ctx.gather(future))
+ finally:
+ _close_runner_context(ctx)
+
+ assert outcomes[0].value == "done"
+ assert len(calls) == 2
+ assert [slot.status for slot in store.call_results] == ["SUCCEEDED"]
+
+
+def
test_exception_whose_message_cannot_be_formatted_is_recorded_by_async_call() ->
(
+ None
+):
+ def call() -> None:
+ raise _UnprintableError
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(_UnprintableError):
+ _run_async(ctx.durable_execute_async(call))
+ finally:
+ _close_runner_context(ctx)
+
+ assert [slot.status for slot in store.call_results] == ["FAILED"]
+
+
+def test_java_interruption_in_gather_leaves_the_slot_pending() -> None:
+ interruption = _java_error(_INTERRUPTED)
+
+ def interrupted() -> None:
+ raise interruption
+
+ def declined() -> None:
+ msg = "payment declined"
+ raise _UnpicklableError(msg)
+
+ def completed() -> str:
+ return "done"
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ with pytest.raises(RuntimeError) as raised:
+ _run_async(
+ ctx.gather(
+ ctx.durable_execute_async(declined),
+ ctx.durable_execute_async(interrupted),
+ ctx.durable_execute_async(completed),
+ )
+ )
+ finally:
+ _close_runner_context(ctx)
+
+ assert raised.value is interruption
+ assert [slot.status for slot in store.call_results] == [
+ "FAILED",
+ "PENDING",
+ "PENDING",
+ ]
+ assert store.current_call_index == 0
+
+
+def test_gather_records_exception_that_cannot_be_pickled() -> None:
+ calls = []
+
+ def declined() -> None:
+ calls.append(1)
+ msg = "payment declined"
+ raise _UnpicklableError(msg)
+
+ store = _FakeJavaRunnerContext()
+ errors = []
+ for _ in range(2):
+ store.current_call_index = 0
+ ctx = _create_runner_context(store)
+ try:
+ outcomes =
_run_async(ctx.gather(ctx.durable_execute_async(declined)))
+ errors.append(outcomes[0].error)
+ finally:
+ _close_runner_context(ctx)
+
+ assert type(errors[0]) is _UnpicklableError
+ assert str(errors[1]) == f"{_UNPICKLABLE_NAME}: payment declined"
+ assert len(calls) == 1
+
+
+def test_interrupted_gather_is_executed_again_when_awaited_again() -> None:
+ calls = []
+
+ def call() -> str:
+ calls.append(1)
+ if len(calls) == 1:
+ raise _java_error(_INTERRUPTED)
+ return "done"
+
+ store = _FakeJavaRunnerContext()
+ ctx = _create_runner_context(store)
+ try:
+ gathered = ctx.gather(ctx.durable_execute_async(call))
+ with pytest.raises(RuntimeError):
+ _run_async(gathered)
+ outcomes = _run_async(gathered)
+ finally:
+ _close_runner_context(ctx)
+
+ assert outcomes[0].value == "done"
+ assert len(calls) == 2
+ assert [slot.status for slot in store.call_results] == ["SUCCEEDED"]
+
+
+def test_inspection_never_replaces_the_failure() -> None:
+ class _Hostile(Exception):
+ @property
+ def args(self) -> tuple:
+ raise KeyboardInterrupt
+
+ assert not is_java_interruption(_Hostile())
+
+
+def test_long_message_is_stored_once() -> None:
+ message = "x" * 450_000
+ error = ValueError(message)
+
+ payload = serialize_durable_exception(error)
+ restored = deserialize_durable_exception(payload)
+
+ assert len(payload) < len(cloudpickle.dumps(error)) + 20 * 1024
+ assert restored.args == (message,)
+
+
+def test_long_message_of_unpicklable_exception_is_truncated_in_the_fallback()
-> None:
+ restored = deserialize_durable_exception(
+ serialize_durable_exception(_UnpicklableError("x" * 20_000))
+ )
+
+ text = str(restored)
+ assert text.startswith(f"{_UNPICKLABLE_NAME}: " + "x" * 16 * 1024)
+ assert text.endswith("... [truncated]")
+ assert len(text) < 17 * 1024
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 f034ae44a..7c272e5eb 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
@@ -26,6 +26,7 @@ import cloudpickle
import pytest
from flink_agents.plan.configuration import AgentConfiguration
+from flink_agents.runtime.durable_exception import
deserialize_durable_exception
from flink_agents.runtime.durable_execution import (
durable_identity_for_call,
)
@@ -352,7 +353,7 @@ def
test_flink_runner_context_sync_reconciler_exception_persists_failure() -> No
assert call_count == 0
assert j_runner_context.operations == ["peek", "finalize"]
assert j_runner_context.call_results[0].status == "FAILED"
- persisted_exception = cloudpickle.loads(
+ persisted_exception = deserialize_durable_exception(
j_runner_context.call_results[0].exception_payload
)
assert isinstance(persisted_exception, ValueError)
@@ -822,7 +823,7 @@ def
test_flink_runner_context_async_reconciler_exception_persists_failure() -> N
assert call_count == 0
assert j_runner_context.operations == ["peek", "finalize"]
assert j_runner_context.call_results[0].status == "FAILED"
- persisted_exception = cloudpickle.loads(
+ persisted_exception = deserialize_durable_exception(
j_runner_context.call_results[0].exception_payload
)
assert isinstance(persisted_exception, RuntimeError)
