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 67c66a829 align action recovery model (#1155)
67c66a829 is described below
commit 67c66a829281793c627b74a1a6f72387ec13c4f2
Author: daken <[email protected]>
AuthorDate: Thu Oct 8 10:08:13 2026 +0800
align action recovery model (#1155)
Generated-by: cursor (deepseek-v4-pro)
Co-authored-by: daken <[email protected]>
---
.../flink/agents/plan/actions/ToolCallAction.java | 4 +-
.../agents/plan/actions/ToolCallActionTest.java | 14 +-
.../flink_agents/plan/actions/tool_call_action.py | 5 +-
.../plan/tests/actions/test_tool_call_action.py | 11 +-
.../actions/test_tool_call_action_subagent.py | 3 +
python/flink_agents/runtime/durable_execution.py | 17 +-
.../flink_agents/runtime/flink_runner_context.py | 47 +++---
.../test_flink_runner_context_reconcilable.py | 94 +++++------
.../agents/runtime/actionstate/CallResult.java | 54 ++-----
.../runtime/context/JavaRunnerContextImpl.java | 21 +--
.../agents/runtime/context/RunnerContextImpl.java | 171 ++++++++-------------
.../runtime/actionstate/ActionStateSerdeTest.java | 22 +--
.../runtime/actionstate/ActionStateTest.java | 66 ++++----
.../agents/runtime/actionstate/CallResultTest.java | 56 +++----
.../context/DurableExecutionContextTest.java | 44 +++---
...vaRunnerContextImplDurableExecuteAsyncTest.java | 20 ++-
.../RunnerContextImplDurableExecuteTest.java | 19 ++-
.../operator/ActionExecutionOperatorTest.java | 11 +-
18 files changed, 286 insertions(+), 393 deletions(-)
diff --git
a/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
b/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
index f49f32847..50aee0990 100644
---
a/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
+++
b/plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java
@@ -61,7 +61,7 @@ import java.util.concurrent.CompletionException;
/** Built-in action for processing tool call. */
public class ToolCallAction {
- static final String TOOL_CALL_DURABLE_ID = "tool-call";
+ static final String TOOL_CALL_DURABLE_ID = "tool-call:";
private static final Logger LOG =
LoggerFactory.getLogger(ToolCallAction.class);
public static Action getToolCallAction() throws Exception {
@@ -218,7 +218,7 @@ public class ToolCallAction {
new DurableCallable<>() {
@Override
public String getId() {
- return TOOL_CALL_DURABLE_ID;
+ return TOOL_CALL_DURABLE_ID + id;
}
@Override
diff --git
a/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
index 9513ba603..c467f47e3 100644
---
a/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
+++
b/plan/src/test/java/org/apache/flink/agents/plan/actions/ToolCallActionTest.java
@@ -480,8 +480,9 @@ public class ToolCallActionTest {
ToolCallAction.processToolRequest(toolRequest("queryOrder", "call-1",
"call-2"), ctx);
- assertThat(ctx.gatherIds).containsExactly(List.of("tool-call",
"tool-call"));
- assertThat(ctx.durableExecuteAsyncIds).containsExactly("tool-call",
"tool-call");
+ assertThat(ctx.gatherIds).containsExactly(List.of("tool-call:call-1",
"tool-call:call-2"));
+ assertThat(ctx.durableExecuteAsyncIds)
+ .containsExactly("tool-call:call-1", "tool-call:call-2");
assertThat(ctx.durableExecuteIds).isEmpty();
ToolResponseEvent response =
ToolResponseEvent.fromEvent(ctx.sentEvents.get(0));
assertThat(response.getResponses().get("call-1").getResult()).isEqualTo("tenant-1:order-1");
@@ -495,7 +496,8 @@ public class ToolCallActionTest {
ToolCallAction.processToolRequest(toolRequest("queryOrder", "call-1",
"call-2"), ctx);
assertThat(ctx.gatherIds).isEmpty();
- assertThat(ctx.durableExecuteAsyncIds).containsExactly("tool-call",
"tool-call");
+ assertThat(ctx.durableExecuteAsyncIds)
+ .containsExactly("tool-call:call-1", "tool-call:call-2");
assertThat(ctx.durableExecuteIds).isEmpty();
}
@@ -507,7 +509,7 @@ public class ToolCallActionTest {
assertThat(ctx.gatherIds).isEmpty();
assertThat(ctx.durableExecuteAsyncIds).isEmpty();
- assertThat(ctx.durableExecuteIds).containsExactly("tool-call",
"tool-call");
+ assertThat(ctx.durableExecuteIds).containsExactly("tool-call:call-1",
"tool-call:call-2");
}
@Test
@@ -517,7 +519,7 @@ public class ToolCallActionTest {
ToolCallAction.processToolRequest(toolRequest("queryOrder"), ctx);
assertThat(ctx.gatherIds).isEmpty();
- assertThat(ctx.durableExecuteAsyncIds).containsExactly("tool-call");
+
assertThat(ctx.durableExecuteAsyncIds).containsExactly("tool-call:call-1");
assertThat(ctx.durableExecuteIds).isEmpty();
}
@@ -543,7 +545,7 @@ public class ToolCallActionTest {
toolCall("queryOrder", "call-2", "order-2"))),
ctx);
- assertThat(ctx.gatherIds).containsExactly(List.of("tool-call",
"tool-call"));
+ assertThat(ctx.gatherIds).containsExactly(List.of("tool-call:call-1",
"tool-call:call-2"));
ToolResponseEvent response =
ToolResponseEvent.fromEvent(ctx.sentEvents.get(0));
assertThat(response.getSuccess()).containsEntry("missing-call", false);
assertThat(response.getError()).containsEntry("missing-call", "missing
resource");
diff --git a/python/flink_agents/plan/actions/tool_call_action.py
b/python/flink_agents/plan/actions/tool_call_action.py
index f68f3c490..74c5d739a 100644
--- a/python/flink_agents/plan/actions/tool_call_action.py
+++ b/python/flink_agents/plan/actions/tool_call_action.py
@@ -315,7 +315,8 @@ async def _execute_parallel(
try:
futures = [
ctx.durable_execute_async(
- execution.func, *execution.args, **execution.kwargs
+ execution.func, *execution.args, **execution.kwargs,
+ durable_id="tool-call:" + execution.id,
)
for execution in tool_executions
]
@@ -359,12 +360,14 @@ async def _execute_sequentially(
execution.func,
*execution.args,
**execution.kwargs,
+ durable_id="tool-call:" + execution.id,
)
else:
response = ctx.durable_execute(
execution.func,
*execution.args,
**execution.kwargs,
+ durable_id="tool-call:" + execution.id,
)
result_observed_at = datetime.now(timezone.utc)
outcome = Outcome.success(response)
diff --git a/python/flink_agents/plan/tests/actions/test_tool_call_action.py
b/python/flink_agents/plan/tests/actions/test_tool_call_action.py
index 9f42b5534..71581c239 100644
--- a/python/flink_agents/plan/tests/actions/test_tool_call_action.py
+++ b/python/flink_agents/plan/tests/actions/test_tool_call_action.py
@@ -68,7 +68,7 @@ def _expected_durable_function_id(
tenant_id: str = "tenant-1",
) -> str:
tool = _query_order_tool()
- function_id, _ = durable_identity_for_call(
+ function_id = durable_identity_for_call(
tool.call,
(),
{"order_id": order_id, "tenant_id": tenant_id},
@@ -138,10 +138,12 @@ class _Context:
def durable_execute(self, func: Any, *args: Any, **kwargs: Any) -> Any:
self.durable_execute_calls.append((func, args, kwargs))
+ kwargs.pop("durable_id", None)
return func(*args, **kwargs)
def durable_execute_async(self, func: Any, *args: Any, **kwargs: Any) ->
Any:
self.durable_execute_async_calls.append((func, args, kwargs))
+ kwargs.pop("durable_id", None)
async def execute() -> Any:
return func(*args, **kwargs)
@@ -404,13 +406,12 @@ def
test_tool_call_action_uses_parallel_batch_for_multiple_tools() -> None:
}
assert response.success == {"call-1": True, "call-2": True}
assert len(ctx.gather_calls) == 1
- expected_id = _expected_durable_function_id("order-call-1")
assert [
- durable_identity_for_call(func, args, kwargs)[0]
+ durable_identity_for_call(func, args, kwargs)
for func, args, kwargs in ctx.gather_calls[0]
] == [
- expected_id,
- expected_id,
+ _expected_durable_function_id("order-call-1"),
+ _expected_durable_function_id("order-call-2"),
]
assert len(ctx.durable_execute_async_calls) == 2
diff --git
a/python/flink_agents/plan/tests/actions/test_tool_call_action_subagent.py
b/python/flink_agents/plan/tests/actions/test_tool_call_action_subagent.py
index 45877c216..3991e0335 100644
--- a/python/flink_agents/plan/tests/actions/test_tool_call_action_subagent.py
+++ b/python/flink_agents/plan/tests/actions/test_tool_call_action_subagent.py
@@ -203,10 +203,13 @@ class _Context(ExecutionReporter):
return registry[name]
def durable_execute(self, func: Any, *args: Any, **kwargs: Any) -> Any:
+ kwargs.pop("durable_id", None)
self.durable_executions += 1
return func(*args, **kwargs)
def durable_execute_async(self, func: Any, *args: Any, **kwargs: Any) ->
Any:
+ kwargs.pop("durable_id", None)
+
# Deferred: the count reflects executions (when the future is awaited,
# directly or via gather), not handle creations, matching the lazy
# durable-future contract.
diff --git a/python/flink_agents/runtime/durable_execution.py
b/python/flink_agents/runtime/durable_execution.py
index a862d02e8..7ddf04e34 100644
--- a/python/flink_agents/runtime/durable_execution.py
+++ b/python/flink_agents/runtime/durable_execution.py
@@ -54,10 +54,21 @@ def durable_identity_for_call(
func: Callable,
args: tuple,
kwargs: dict | None,
-) -> tuple[str, str]:
- """Return the durable journal identity for a single callable invocation."""
+) -> str:
+ """Return the durable journal identity for a single callable invocation.
+
+ When the callable carries an explicit id attached by
:func:`with_durable_id`,
+ that id is returned directly. Otherwise the identity is derived as
+ ``module.qualname`` concatenated with a stable digest of the serialized
+ arguments.
+ """
+ explicit_id = get_durable_id(func)
+ if explicit_id is not None:
+ return explicit_id
call_kwargs = kwargs or {}
- return _compute_function_id(func), _compute_args_digest(args, call_kwargs)
+ function_id = _compute_function_id(func)
+ args_digest = _compute_args_digest(args, call_kwargs)
+ return f"{function_id}::{args_digest}"
def _compute_function_id(func: Callable) -> str:
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index 14c33fccd..c385b7b82 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -46,8 +46,6 @@ from flink_agents.api.runner_context import (
)
from flink_agents.api.trace import ExecutionReporter
from flink_agents.runtime.durable_execution import (
- _compute_args_digest,
- _compute_function_id,
_validate_reconciler_callable,
durable_identity_for_call,
with_durable_id,
@@ -118,7 +116,6 @@ class _AsyncExecutionResult:
@dataclass(frozen=True)
class _PersistedCallResult:
function_id: str
- args_digest: str
status: str
result_payload: bytes | None
exception_payload: bytes | None
@@ -892,12 +889,12 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
A tuple of (is_hit, result_or_exception). If is_hit is True,
the second element is the cached result or an exception to
re-raise.
"""
- function_id, args_digest = durable_identity_for_call(func, args,
kwargs)
+ function_id = durable_identity_for_call(func, args, kwargs)
cached_exception: BaseException | None = None
try:
cached =
self._j_runner_context.matchNextOrClearSubsequentCallResult(
- function_id, args_digest
+ function_id
)
if cached is not None:
is_hit, result_payload, exception_payload = cached
@@ -945,14 +942,14 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
exception : BaseException | None
The exception raised by the function (None if successful).
"""
- function_id, args_digest = durable_identity_for_call(func, args,
kwargs)
+ 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
self._j_runner_context.recordCallCompletion(
- function_id, args_digest, result_payload, exception_payload
+ function_id, result_payload, exception_payload
)
except Exception as e:
# If Java method doesn't exist, silently ignore
@@ -973,10 +970,9 @@ class FlinkRunnerContext(RunnerContext, ExecutionReporter):
if current is None:
return None
- function_id, args_digest, status, result_payload, exception_payload =
current
+ function_id, status, result_payload, exception_payload = current
return _PersistedCallResult(
function_id=function_id,
- args_digest=args_digest,
status=status,
result_payload=bytes(result_payload)
if result_payload is not None
@@ -991,10 +987,9 @@ class FlinkRunnerContext(RunnerContext, ExecutionReporter):
if current is None:
return None
- function_id, args_digest, status, result_payload, exception_payload =
current
+ function_id, status, result_payload, exception_payload = current
return _PersistedCallResult(
function_id=function_id,
- args_digest=args_digest,
status=status,
result_payload=bytes(result_payload)
if result_payload is not None
@@ -1006,8 +1001,7 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
def _append_pending_call(self, func: Callable, args: tuple, kwargs: dict)
-> None:
self._j_runner_context.appendPendingCall(
- _compute_function_id(func),
- _compute_args_digest(args, kwargs),
+ durable_identity_for_call(func, args, kwargs),
)
def _finalize_current_call(
@@ -1018,14 +1012,13 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
result: Any,
exception: BaseException | None,
) -> None:
- function_id, args_digest = durable_identity_for_call(func, args,
kwargs)
+ function_id = durable_identity_for_call(func, args, kwargs)
result_payload, exception_payload = self._serialize_call_payloads(
result,
exception,
)
self._j_runner_context.finalizeCurrentCall(
function_id,
- args_digest,
result_payload,
exception_payload,
)
@@ -1047,7 +1040,7 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
reconciler: Callable[[], Any],
kwargs: dict,
) -> _ReconcilerExecutionPlan:
- function_id, args_digest = durable_identity_for_call(func, args,
kwargs)
+ function_id = durable_identity_for_call(func, args, kwargs)
current = self._peek_current_call_result()
durable_call = partial(func, *args, **kwargs)
@@ -1058,7 +1051,7 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
needs_append_pending=True,
)
- if current.function_id != function_id or current.args_digest !=
args_digest:
+ if current.function_id != function_id:
return _ReconcilerExecutionPlan(
"execute",
callable=durable_call,
@@ -1092,12 +1085,11 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
args: tuple,
kwargs: dict,
) -> bool:
- function_id, args_digest = durable_identity_for_call(func, args,
kwargs)
+ function_id = durable_identity_for_call(func, args, kwargs)
current = self._peek_current_call_result()
return (
current is not None
and current.function_id == function_id
- and current.args_digest == args_digest
and current.status == "PENDING"
)
@@ -1199,14 +1191,14 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
return wrapped_func
- def _durable_identity(self, call: _DurableCall) -> tuple[str, str]:
+ def _durable_identity(self, call: _DurableCall) -> str:
return durable_identity_for_call(call.func, call.args, call.kwargs)
def _call_matches(
self, current: _PersistedCallResult, call: _DurableCall
) -> bool:
- function_id, args_digest = self._durable_identity(call)
- return current.function_id == function_id and current.args_digest ==
args_digest
+ function_id = self._durable_identity(call)
+ return current.function_id == function_id
def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
try:
@@ -1230,7 +1222,7 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
execution_start = -1
for index, call in enumerate(calls):
- function_id, args_digest = self._durable_identity(call)
+ function_id = self._durable_identity(call)
current = self._read_call_result_at(base + index)
if current is None:
needs_reservation = True
@@ -1269,12 +1261,10 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
if needs_reservation:
function_ids = []
- args_digests = []
for call in calls[execution_start:]:
- function_id, args_digest = self._durable_identity(call)
+ function_id = self._durable_identity(call)
function_ids.append(function_id)
- args_digests.append(args_digest)
- self._j_runner_context.reservePendingBatch(function_ids,
args_digests)
+ self._j_runner_context.reservePendingBatch(function_ids)
return _BatchExecutionPlan(
outcomes=outcomes,
@@ -1296,7 +1286,7 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
zip(plan.suppliers, executed, strict=True)
):
call = calls[call_index]
- function_id, args_digest = self._durable_identity(call)
+ function_id = self._durable_identity(call)
if not started[i]:
outcomes[call_index] = outcome
continue
@@ -1308,7 +1298,6 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
self._j_runner_context.finalizeCallAt(
base + call_index,
function_id,
- args_digest,
result_payload,
exception_payload,
)
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 b3a235f4a..f034ae44a 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
@@ -27,8 +27,6 @@ import pytest
from flink_agents.plan.configuration import AgentConfiguration
from flink_agents.runtime.durable_execution import (
- _compute_args_digest,
- _compute_function_id,
durable_identity_for_call,
)
from flink_agents.runtime.flink_runner_context import FlinkRunnerContext
@@ -37,7 +35,6 @@ from flink_agents.runtime.flink_runner_context import
FlinkRunnerContext
@dataclass
class _StoredCallResult:
function_id: str
- args_digest: str
status: str
result_payload: bytes | None = None
exception_payload: bytes | None = None
@@ -49,10 +46,9 @@ def _stored_call(
status: str,
**kwargs: Any,
) -> _StoredCallResult:
- function_id, args_digest = durable_identity_for_call(func, args, kwargs or
None)
+ function_id = durable_identity_for_call(func, args, kwargs or None)
return _StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
status=status,
)
@@ -72,7 +68,6 @@ class _FakeJavaRunnerContext:
current = self.call_results[index]
return [
current.function_id,
- current.args_digest,
current.status,
current.result_payload,
current.exception_payload,
@@ -85,7 +80,6 @@ class _FakeJavaRunnerContext:
current = self.call_results[self.current_call_index]
return [
current.function_id,
- current.args_digest,
current.status,
current.result_payload,
current.exception_payload,
@@ -93,15 +87,12 @@ class _FakeJavaRunnerContext:
return None
def matchNextOrClearSubsequentCallResult(
- self, function_id: str, args_digest: str
+ self, function_id: str
) -> list[Any] | None:
self.operations.append("match")
if self.current_call_index < len(self.call_results):
current = self.call_results[self.current_call_index]
- if (
- current.function_id == function_id
- and current.args_digest == args_digest
- ):
+ if current.function_id == function_id:
if current.status == "PENDING":
return None
self.current_call_index += 1
@@ -112,7 +103,6 @@ class _FakeJavaRunnerContext:
def recordCallCompletion(
self,
function_id: str,
- args_digest: str,
result_payload: bytes | None,
exception_payload: bytes | None,
) -> None:
@@ -121,7 +111,7 @@ class _FakeJavaRunnerContext:
self.call_results.append(
_StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status=status,
result_payload=result_payload,
exception_payload=exception_payload,
@@ -129,12 +119,12 @@ class _FakeJavaRunnerContext:
)
self.current_call_index += 1
- def appendPendingCall(self, function_id: str, args_digest: str) -> None:
+ def appendPendingCall(self, function_id: str) -> None:
self.operations.append("append_pending")
self.call_results.append(
_StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status="PENDING",
)
)
@@ -142,7 +132,6 @@ class _FakeJavaRunnerContext:
def finalizeCurrentCall(
self,
function_id: str,
- args_digest: str,
result_payload: bytes | None,
exception_payload: bytes | None,
) -> None:
@@ -150,10 +139,9 @@ class _FakeJavaRunnerContext:
current = self.call_results[self.current_call_index]
assert current.status == "PENDING"
assert current.function_id == function_id
- assert current.args_digest == args_digest
self.call_results[self.current_call_index] = _StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status="FAILED" if exception_payload is not None else "SUCCEEDED",
result_payload=result_payload,
exception_payload=exception_payload,
@@ -165,14 +153,14 @@ class _FakeJavaRunnerContext:
self.call_results = self.call_results[: self.current_call_index]
def reservePendingBatch(
- self, function_ids: list[str], args_digests: list[str]
+ self, function_ids: list[str]
) -> None:
self.operations.append(f"reserve:{len(function_ids)}")
- for function_id, digest in zip(function_ids, args_digests,
strict=True):
+ for function_id in function_ids:
self.call_results.append(
_StoredCallResult(
function_id=function_id,
- args_digest=digest,
+
status="PENDING",
)
)
@@ -181,7 +169,6 @@ class _FakeJavaRunnerContext:
self,
index: int,
function_id: str,
- args_digest: str,
result_payload: bytes | None,
exception_payload: bytes | None,
) -> None:
@@ -189,10 +176,9 @@ class _FakeJavaRunnerContext:
current = self.call_results[index]
assert current.status == "PENDING"
assert current.function_id == function_id
- assert current.args_digest == args_digest
self.call_results[index] = _StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status="FAILED" if exception_payload is not None else "SUCCEEDED",
result_payload=result_payload,
exception_payload=exception_payload,
@@ -277,8 +263,8 @@ def _preload_pending(
) -> None:
j_runner_context.call_results.append(
_StoredCallResult(
- function_id=_compute_function_id(func),
- args_digest=_compute_args_digest(args, kwargs),
+ function_id=durable_identity_for_call(func, args, kwargs or None),
+
status="PENDING",
)
)
@@ -381,13 +367,12 @@ def
test_flink_runner_context_sync_reconciler_mismatch_clears_and_executes() ->
j_runner_context.call_results.extend(
[
_StoredCallResult(
- function_id=_compute_function_id(_call_value),
- args_digest=_compute_args_digest(("other-order",), {}),
+ function_id="wrong.expected.function_id",
+
status="PENDING",
),
_StoredCallResult(
- function_id="stale.function",
- args_digest="stale-args",
+ function_id="stale.different_function",
status="SUCCEEDED",
result_payload=stale_result_payload,
),
@@ -415,11 +400,8 @@ def
test_flink_runner_context_sync_reconciler_mismatch_clears_and_executes() ->
"finalize",
]
assert len(j_runner_context.call_results) == 1
- assert j_runner_context.call_results[0].function_id ==
_compute_function_id(
- _call_value
- )
- assert j_runner_context.call_results[0].args_digest ==
_compute_args_digest(
- ("order-1",), {}
+ assert j_runner_context.call_results[0].function_id ==
durable_identity_for_call(
+ _call_value, ("order-1",), None
)
assert j_runner_context.call_results[0].status == "SUCCEEDED"
@@ -442,17 +424,17 @@ def
test_flink_runner_context_durable_execute_reexecutes_pending_after_batch_res
second_call_count += 1
return "two"
- first_id, first_digest = durable_identity_for_call(first_call, (), None)
- second_id, second_digest = durable_identity_for_call(second_call, (), None)
+ first_id = durable_identity_for_call(first_call, (), None)
+ second_id = durable_identity_for_call(second_call, (), None)
j_runner_context.call_results = [
_StoredCallResult(
function_id=first_id,
- args_digest=first_digest,
+
status="PENDING",
),
_StoredCallResult(
function_id=second_id,
- args_digest=second_digest,
+
status="PENDING",
),
]
@@ -486,11 +468,11 @@ def
test_flink_runner_context_durable_execute_async_reexecutes_pending_after_bat
call_count += 1
return f"call:{value}"
- function_id, args_digest = durable_identity_for_call(tracked_call,
("order-1",), None)
+ function_id = durable_identity_for_call(tracked_call, ("order-1",), None)
j_runner_context.call_results = [
_StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status="PENDING",
),
]
@@ -978,14 +960,14 @@ def
test_flink_runner_context_gather_recovers_partial_batch() -> None:
call_count += 1
return _call_value(value)
- cached_function_id, cached_digest = durable_identity_for_call(
+ cached_function_id = durable_identity_for_call(
tracked_call, ("one",), None
)
j_runner_context.call_results.extend(
[
_StoredCallResult(
function_id=cached_function_id,
- args_digest=cached_digest,
+
status="SUCCEEDED",
result_payload=cloudpickle.dumps("cached-one"),
),
@@ -1012,13 +994,13 @@ def
test_flink_runner_context_gather_recovers_partial_batch() -> None:
def test_flink_runner_context_gather_returns_cached_failure() -> None:
j_runner_context = _FakeJavaRunnerContext()
- cached_function_id, cached_digest = durable_identity_for_call(
+ cached_function_id = durable_identity_for_call(
_call_value, ("one",), None
)
j_runner_context.call_results.append(
_StoredCallResult(
function_id=cached_function_id,
- args_digest=cached_digest,
+
status="FAILED",
exception_payload=cloudpickle.dumps(ValueError("cached failure")),
)
@@ -1206,11 +1188,11 @@ def
test_flink_runner_context_gather_returns_deserialize_failure_as_outcome() ->
return "should-not-run"
func = should_not_run
- function_id, args_digest = durable_identity_for_call(func, (), None)
+ function_id = durable_identity_for_call(func, (), None)
j_runner_context.call_results.append(
_StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status="SUCCEEDED",
result_payload=b"not-valid-pickle",
)
@@ -1241,11 +1223,11 @@ def
test_flink_runner_context_gather_reconciles_pending_slot() -> None:
reconcile_count += 1
return "recovered"
- function_id, args_digest = durable_identity_for_call(tracked_call, (),
None)
+ function_id = durable_identity_for_call(tracked_call, (), None)
j_runner_context.call_results.append(
_StoredCallResult(
function_id=function_id,
- args_digest=args_digest,
+
status="PENDING",
)
)
@@ -1275,28 +1257,28 @@ def
test_flink_runner_context_gather_recovers_three_slot_partial_batch() -> None
call_count += 1
return _call_value(value)
- first_id, first_digest = durable_identity_for_call(tracked_call, ("one",),
None)
- second_id, second_digest = durable_identity_for_call(tracked_call,
("two",), None)
- third_id, third_digest = durable_identity_for_call(
+ first_id = durable_identity_for_call(tracked_call, ("one",), None)
+ second_id = durable_identity_for_call(tracked_call, ("two",), None)
+ third_id = durable_identity_for_call(
tracked_call, ("three",), None
)
j_runner_context.call_results.extend(
[
_StoredCallResult(
function_id=first_id,
- args_digest=first_digest,
+
status="SUCCEEDED",
result_payload=cloudpickle.dumps("cached-one"),
),
_StoredCallResult(
function_id=second_id,
- args_digest=second_digest,
+
status="SUCCEEDED",
result_payload=cloudpickle.dumps("cached-two"),
),
_StoredCallResult(
function_id=third_id,
- args_digest=third_digest,
+
status="PENDING",
),
]
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/CallResult.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/CallResult.java
index 62906d815..adc0d57e6 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/CallResult.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/CallResult.java
@@ -18,6 +18,7 @@
package org.apache.flink.agents.runtime.actionstate;
import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.Arrays;
@@ -33,6 +34,7 @@ import java.util.Objects;
* <p>During recovery, the success or failure of the original call is
determined by checking whether
* {@code exceptionPayload} is null.
*/
+@JsonIgnoreProperties(ignoreUnknown = true)
public class CallResult {
/** Persisted status of the durable call. */
@@ -42,12 +44,9 @@ public class CallResult {
FAILED
}
- /** Function identifier: module+qualname for Python, or method signature
for Java. */
+ /** Function identifier that uniquely identifies the durable call for
recovery matching. */
private final String functionId;
- /** Stable digest of the serialized arguments for validation during
recovery. */
- private final String argsDigest;
-
/** Serialized return value of the function call (null if the call threw
an exception). */
private final byte[] resultPayload;
@@ -61,7 +60,6 @@ public class CallResult {
/** Default constructor for deserialization. */
public CallResult() {
this.functionId = null;
- this.argsDigest = null;
this.resultPayload = null;
this.exceptionPayload = null;
this.status = null;
@@ -71,12 +69,10 @@ public class CallResult {
* Constructs a CallResult for a successful function call.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @param resultPayload the serialized return value
*/
- public CallResult(String functionId, String argsDigest, byte[]
resultPayload) {
+ public CallResult(String functionId, byte[] resultPayload) {
this.functionId = functionId;
- this.argsDigest = argsDigest;
this.resultPayload = resultPayload;
this.exceptionPayload = null;
this.status = Status.SUCCEEDED;
@@ -86,15 +82,12 @@ public class CallResult {
* Constructs a CallResult with explicit result and exception payloads.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @param resultPayload the serialized return value (null if exception
occurred)
* @param exceptionPayload the serialized exception (null if call
succeeded)
*/
- public CallResult(
- String functionId, String argsDigest, byte[] resultPayload, byte[]
exceptionPayload) {
+ public CallResult(String functionId, byte[] resultPayload, byte[]
exceptionPayload) {
this(
functionId,
- argsDigest,
resultPayload,
exceptionPayload,
exceptionPayload == null ? Status.SUCCEEDED : Status.FAILED);
@@ -104,19 +97,13 @@ public class CallResult {
* Constructs a CallResult with explicit result, exception payloads, and
status.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @param resultPayload the serialized return value (null if exception
occurred or pending)
* @param exceptionPayload the serialized exception (null if call
succeeded or pending)
* @param status the persisted call status
*/
private CallResult(
- String functionId,
- String argsDigest,
- byte[] resultPayload,
- byte[] exceptionPayload,
- Status status) {
+ String functionId, byte[] resultPayload, byte[] exceptionPayload,
Status status) {
this.functionId = functionId;
- this.argsDigest = argsDigest;
this.resultPayload = resultPayload;
this.exceptionPayload = exceptionPayload;
this.status = status;
@@ -127,21 +114,16 @@ public class CallResult {
* persisted.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @return a new CallResult representing a pending call
*/
- public static CallResult pending(String functionId, String argsDigest) {
- return new CallResult(functionId, argsDigest, null, null,
Status.PENDING);
+ public static CallResult pending(String functionId) {
+ return new CallResult(functionId, null, null, Status.PENDING);
}
public String getFunctionId() {
return functionId;
}
- public String getArgsDigest() {
- return argsDigest;
- }
-
public byte[] getResultPayload() {
return resultPayload;
}
@@ -151,15 +133,13 @@ public class CallResult {
}
/**
- * Validates if this CallResult matches the given function identifier and
arguments digest.
+ * Validates if this CallResult matches the given function identifier.
*
* @param functionId the function identifier to match
- * @param argsDigest the arguments digest to match
- * @return true if both functionId and argsDigest match, false otherwise
+ * @return true if functionId matches, false otherwise
*/
- public boolean matches(String functionId, String argsDigest) {
- return Objects.equals(this.functionId, functionId)
- && Objects.equals(this.argsDigest, argsDigest);
+ public boolean matches(String functionId) {
+ return Objects.equals(this.functionId, functionId);
}
/**
@@ -190,8 +170,8 @@ public class CallResult {
* <p>Used by backward-compatibility tests for legacy serialized state.
*/
static CallResult ofNullStatus(
- String functionId, String argsDigest, byte[] resultPayload, byte[]
exceptionPayload) {
- return new CallResult(functionId, argsDigest, resultPayload,
exceptionPayload, null);
+ String functionId, byte[] resultPayload, byte[] exceptionPayload) {
+ return new CallResult(functionId, resultPayload, exceptionPayload,
null);
}
/**
@@ -218,7 +198,6 @@ public class CallResult {
}
CallResult that = (CallResult) o;
return Objects.equals(functionId, that.functionId)
- && Objects.equals(argsDigest, that.argsDigest)
&& Arrays.equals(resultPayload, that.resultPayload)
&& Arrays.equals(exceptionPayload, that.exceptionPayload)
&& status == that.status;
@@ -226,7 +205,7 @@ public class CallResult {
@Override
public int hashCode() {
- int result = Objects.hash(functionId, argsDigest, status);
+ int result = Objects.hash(functionId, status);
result = 31 * result + Arrays.hashCode(resultPayload);
result = 31 * result + Arrays.hashCode(exceptionPayload);
return result;
@@ -238,9 +217,6 @@ public class CallResult {
+ "functionId='"
+ functionId
+ '\''
- + ", argsDigest='"
- + argsDigest
- + '\''
+ ", resultPayload="
+ (resultPayload != null ? resultPayload.length + " bytes" :
"null")
+ ", exceptionPayload="
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 5d5dbeca1..e7dfe2c4c 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
@@ -94,21 +94,20 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
return executeAllWithoutDurableState(callables);
}
- String argsDigest = "";
int base = durableExecutionContext.getCurrentCallIndex();
- BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base,
argsDigest);
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base);
- reservePendingBatchIfNeeded(callables, argsDigest, plan);
+ reservePendingBatchIfNeeded(callables, plan);
BatchExecutionResult<T> executed =
executeOutcomeSuppliers(plan.suppliers);
- finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+ finalizeExecutedOutcomes(callables, base, plan, executed);
advanceCallIndexBy(callables.size());
return plan.outcomes;
}
private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
- List<DurableCallable<T>> callables, int base, String argsDigest)
throws Exception {
+ List<DurableCallable<T>> callables, int base) throws Exception {
BatchExecutionPlan<T> plan = new
BatchExecutionPlan<>(callables.size());
for (int i = 0; i < callables.size(); i++) {
DurableCallable<T> callable = callables.get(i);
@@ -118,7 +117,7 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
addExecutableCall(plan, i, callable::call);
continue;
}
- if (!current.matches(callable.getId(), argsDigest)) {
+ if (!current.matches(callable.getId())) {
clearCallResultsFromAndPersist(base + i);
plan.needsReservation = true;
plan.executionStart = i;
@@ -134,7 +133,7 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
} else {
plan.outcomes.add(
readTerminalOutcomeAt(
- base + i, callable.getId(), argsDigest,
callable.getResultClass()));
+ base + i, callable.getId(),
callable.getResultClass()));
}
}
return plan;
@@ -163,24 +162,21 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
}
private <T> void reservePendingBatchIfNeeded(
- List<DurableCallable<T>> callables, String argsDigest,
BatchExecutionPlan<T> plan) {
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan) {
if (!plan.needsReservation) {
return;
}
List<String> ids = new ArrayList<>();
- List<String> argsDigests = new ArrayList<>();
for (DurableCallable<T> callable :
callables.subList(plan.executionStart, callables.size())) {
ids.add(callable.getId());
- argsDigests.add(argsDigest);
}
- reservePendingBatch(ids, argsDigests);
+ reservePendingBatch(ids);
}
private <T> void finalizeExecutedOutcomes(
List<DurableCallable<T>> callables,
int base,
- String argsDigest,
BatchExecutionPlan<T> plan,
BatchExecutionResult<T> executed)
throws Exception {
@@ -205,7 +201,6 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
finalizeCallAt(
base + callIndex,
callable.getId(),
- argsDigest,
serializeDurableResult(outcome.getValue()),
serializeDurableException(outcome.getError()));
} catch (Exception e) {
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 e09d50b8e..691be4113 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
@@ -612,16 +612,14 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
protected <T> T durableExecuteCompletionOnly(
DurableCallable<T> durableCallable, Callable<T> executionCallable)
throws Exception {
String functionId = durableCallable.getId();
- // argsDigest is empty because DurableCallable encapsulates all
arguments internally
- String argsDigest = "";
CallResult current = getCurrentCallResult();
- if (current != null && current.matches(functionId, argsDigest) &&
current.isPending()) {
- return executeAndFinalizeCurrentCall(functionId, argsDigest,
executionCallable);
+ if (current != null && current.matches(functionId) &&
current.isPending()) {
+ return executeAndFinalizeCurrentCall(functionId,
executionCallable);
}
Optional<Outcome<T>> cachedResult =
- tryGetCachedResult(functionId, argsDigest,
durableCallable.getResultClass());
+ tryGetCachedResult(functionId,
durableCallable.getResultClass());
if (cachedResult.isPresent()) {
return cachedResult.get().getValue();
}
@@ -640,7 +638,7 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
exception = e;
}
- recordDurableCompletion(functionId, argsDigest, result, exception);
+ recordDurableCompletion(functionId, result, exception);
if (exception != null) {
throw exception;
@@ -745,15 +743,13 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
* <p>This method delegates to the {@link DurableExecutionContext} if
present.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @return array containing [isHit (boolean), resultPayload (byte[]),
exceptionPayload
* (byte[])], or null if miss or durable execution is not enabled
*/
- public Object[] matchNextOrClearSubsequentCallResult(String functionId,
String argsDigest) {
+ public Object[] matchNextOrClearSubsequentCallResult(String functionId) {
mailboxThreadChecker.run();
if (durableExecutionContext != null) {
- return
durableExecutionContext.matchNextOrClearSubsequentCallResult(
- functionId, argsDigest);
+ return
durableExecutionContext.matchNextOrClearSubsequentCallResult(functionId);
}
return null;
}
@@ -764,54 +760,49 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
* <p>This method delegates to the {@link DurableExecutionContext} if
present.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @param resultPayload the serialized result (null if exception)
* @param exceptionPayload the serialized exception (null if success)
*/
public void recordCallCompletion(
- String functionId, String argsDigest, byte[] resultPayload, byte[]
exceptionPayload) {
+ String functionId, byte[] resultPayload, byte[] exceptionPayload) {
mailboxThreadChecker.run();
if (durableExecutionContext != null) {
durableExecutionContext.recordCallCompletion(
- functionId, argsDigest, resultPayload, exceptionPayload);
+ functionId, resultPayload, exceptionPayload);
}
}
/** Appends a pending durable call slot at the current call index. */
- public void appendPendingCall(String functionId, String argsDigest) {
+ public void appendPendingCall(String functionId) {
mailboxThreadChecker.run();
if (durableExecutionContext != null) {
- durableExecutionContext.appendPendingCall(functionId, argsDigest);
+ durableExecutionContext.appendPendingCall(functionId);
}
}
- public void reservePendingBatch(List<String> functionIds, List<String>
argsDigests) {
+ public void reservePendingBatch(List<String> functionIds) {
mailboxThreadChecker.run();
if (durableExecutionContext != null && !functionIds.isEmpty()) {
- durableExecutionContext.reservePendingBatch(functionIds,
argsDigests);
+ durableExecutionContext.reservePendingBatch(functionIds);
}
}
/** Finalizes the pending durable call slot at the current call index. */
public void finalizeCurrentCall(
- String functionId, String argsDigest, byte[] resultPayload, byte[]
exceptionPayload) {
+ String functionId, byte[] resultPayload, byte[] exceptionPayload) {
mailboxThreadChecker.run();
if (durableExecutionContext != null) {
durableExecutionContext.finalizeCurrentCall(
- functionId, argsDigest, resultPayload, exceptionPayload);
+ functionId, resultPayload, exceptionPayload);
}
}
public void finalizeCallAt(
- int index,
- String functionId,
- String argsDigest,
- byte[] resultPayload,
- byte[] exceptionPayload) {
+ int index, String functionId, byte[] resultPayload, byte[]
exceptionPayload) {
mailboxThreadChecker.run();
if (durableExecutionContext != null) {
durableExecutionContext.finalizeCallAt(
- index, functionId, argsDigest, resultPayload,
exceptionPayload);
+ index, functionId, resultPayload, exceptionPayload);
}
}
@@ -854,7 +845,6 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
}
return new Object[] {
current.getFunctionId(),
- current.getArgsDigest(),
current.isPending() ? "PENDING" : current.isFailure() ? "FAILED" :
"SUCCEEDED",
current.getResultPayload(),
current.getExceptionPayload()
@@ -873,15 +863,14 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
}
protected <T> Outcome<T> readTerminalOutcomeAt(
- int index, String functionId, String argsDigest, Class<T>
resultClass)
- throws Exception {
+ int index, String functionId, Class<T> resultClass) throws
Exception {
CallResult callResult = getCallResultAt(index);
if (callResult == null || callResult.isPending()) {
throw new IllegalStateException(
String.format(
"Expected a terminal durable call result at index
%s for "
- + "functionId=%s, argsDigest=%s",
- index, functionId, argsDigest));
+ + "functionId=%s",
+ index, functionId));
}
try {
if (callResult.getExceptionPayload() != null) {
@@ -922,9 +911,9 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
* 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);
+ protected <T> Optional<Outcome<T>> tryGetCachedResult(String functionId,
Class<T> resultClass)
+ throws Exception {
+ Object[] cached = matchNextOrClearSubsequentCallResult(functionId);
if (cached != null && (Boolean) cached[0]) {
byte[] resultPayload = (byte[]) cached[1];
byte[] exceptionPayload = (byte[]) cached[2];
@@ -943,12 +932,11 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
return Optional.empty();
}
- protected void recordDurableCompletion(
- String functionId, String argsDigest, Object result, Exception
exception)
+ protected void recordDurableCompletion(String functionId, Object result,
Exception exception)
throws Exception {
byte[] resultPayload = serializeDurableResult(result);
byte[] exceptionPayload = serializeDurableException(exception);
- recordCallCompletion(functionId, argsDigest, resultPayload,
exceptionPayload);
+ recordCallCompletion(functionId, resultPayload, exceptionPayload);
}
/**
@@ -967,41 +955,40 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
Callable<T> executionCallable)
throws Exception {
String functionId = durableCallable.getId();
- String argsDigest = "";
Preconditions.checkState(
durableExecutionContext != null, "durableExecutionContext must
not be null");
CallResult current = getCurrentCallResult();
if (current == null) {
- appendPendingCall(functionId, argsDigest);
- return executeAndFinalizeCurrentCall(functionId, argsDigest,
executionCallable);
+ appendPendingCall(functionId);
+ return executeAndFinalizeCurrentCall(functionId,
executionCallable);
}
- if (!current.matches(functionId, argsDigest)) {
+ if (!current.matches(functionId)) {
clearCallResultsFromCurrentIndexAndPersist();
- appendPendingCall(functionId, argsDigest);
- return executeAndFinalizeCurrentCall(functionId, argsDigest,
executionCallable);
+ appendPendingCall(functionId);
+ return executeAndFinalizeCurrentCall(functionId,
executionCallable);
}
if (!current.isPending()) {
Optional<Outcome<T>> cachedResult =
- tryGetCachedResult(functionId, argsDigest,
durableCallable.getResultClass());
+ tryGetCachedResult(functionId,
durableCallable.getResultClass());
if (cachedResult.isPresent()) {
return cachedResult.get().getValue();
}
throw new IllegalStateException(
String.format(
"Expected a terminal durable call result at index
%s for "
- + "functionId=%s, argsDigest=%s",
- durableExecutionContext.getCurrentCallIndex(),
functionId, argsDigest));
+ + "functionId=%s",
+ durableExecutionContext.getCurrentCallIndex(),
functionId));
}
- return executeAndFinalizeCurrentCall(functionId, argsDigest,
reconcileCallable);
+ return executeAndFinalizeCurrentCall(functionId, reconcileCallable);
}
- protected <T> T executeAndFinalizeCurrentCall(
- String functionId, String argsDigest, Callable<T> callSupplier)
throws Exception {
+ protected <T> T executeAndFinalizeCurrentCall(String functionId,
Callable<T> callSupplier)
+ throws Exception {
T result = null;
Exception exception = null;
try {
@@ -1017,10 +1004,7 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
}
finalizeCurrentCall(
- functionId,
- argsDigest,
- serializeDurableResult(result),
- serializeDurableException(exception));
+ functionId, serializeDurableResult(result),
serializeDurableException(exception));
if (exception != null) {
throw exception;
@@ -1118,28 +1102,25 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
* detected.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @return array containing [isHit, resultPayload, exceptionPayload],
or null if miss
*/
- public Object[] matchNextOrClearSubsequentCallResult(String
functionId, String argsDigest) {
+ public Object[] matchNextOrClearSubsequentCallResult(String
functionId) {
if (currentCallIndex < recoveryCallResults.size()) {
CallResult result = recoveryCallResults.get(currentCallIndex);
- if (result.matches(functionId, argsDigest)) {
+ if (result.matches(functionId)) {
if (result.isPending()) {
LOG.debug(
"Pending CallResult at index {} treated as
cache miss: "
- + "functionId={}, argsDigest={}",
+ + "functionId={}",
currentCallIndex,
- functionId,
- argsDigest);
+ functionId);
return null;
}
LOG.debug(
- "CallResult hit at index {}: functionId={},
argsDigest={}",
+ "CallResult hit at index {}: functionId={}",
currentCallIndex,
- functionId,
- argsDigest);
+ functionId);
currentCallIndex++;
return new Object[] {
true, result.getResultPayload(),
result.getExceptionPayload()
@@ -1147,13 +1128,11 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
} else {
LOG.warn(
"Non-deterministic call detected at index {}:
expected functionId={}, "
- + "argsDigest={}, but got functionId={},
argsDigest={}. "
+ + "but got functionId={}. "
+ "Clearing subsequent results.",
currentCallIndex,
result.getFunctionId(),
- result.getArgsDigest(),
- functionId,
- argsDigest);
+ functionId);
clearCallResultsFromCurrentIndex();
}
}
@@ -1164,27 +1143,21 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
* Records a completed call and persists the ActionState.
*
* @param functionId the function identifier
- * @param argsDigest the digest of serialized arguments
* @param resultPayload the serialized result (null if exception)
* @param exceptionPayload the serialized exception (null if success)
*/
public void recordCallCompletion(
- String functionId,
- String argsDigest,
- byte[] resultPayload,
- byte[] exceptionPayload) {
- CallResult callResult =
- new CallResult(functionId, argsDigest, resultPayload,
exceptionPayload);
+ String functionId, byte[] resultPayload, byte[]
exceptionPayload) {
+ CallResult callResult = new CallResult(functionId, resultPayload,
exceptionPayload);
actionState.addCallResult(callResult);
recoveryCallResults.add(callResult);
persistActionState();
LOG.debug(
- "Recorded and persisted CallResult at index {}:
functionId={}, argsDigest={}",
+ "Recorded and persisted CallResult at index {}:
functionId={}",
currentCallIndex,
- functionId,
- argsDigest);
+ functionId);
currentCallIndex++;
}
@@ -1195,7 +1168,7 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
* <p>This reserves the current slot for a reconcilable durable call
but does not advance
* {@code currentCallIndex}.
*/
- public void appendPendingCall(String functionId, String argsDigest) {
+ public void appendPendingCall(String functionId) {
if (currentCallIndex != recoveryCallResults.size()) {
throw new IllegalStateException(
String.format(
@@ -1204,28 +1177,20 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
currentCallIndex));
}
- CallResult pending = CallResult.pending(functionId, argsDigest);
+ CallResult pending = CallResult.pending(functionId);
actionState.addCallResult(pending);
recoveryCallResults.add(pending);
persistActionState();
LOG.debug(
- "Recorded and persisted pending CallResult at index {}:
functionId={}, "
- + "argsDigest={}",
+ "Recorded and persisted pending CallResult at index {}:
functionId={}",
currentCallIndex,
- functionId,
- argsDigest);
+ functionId);
}
- public void reservePendingBatch(List<String> functionIds, List<String>
argsDigests) {
- if (functionIds.size() != argsDigests.size()) {
- throw new IllegalArgumentException(
- String.format(
- "functionIds size (%s) must match argsDigests
size (%s)",
- functionIds.size(), argsDigests.size()));
- }
+ public void reservePendingBatch(List<String> functionIds) {
for (int i = 0; i < functionIds.size(); i++) {
- CallResult pending = CallResult.pending(functionIds.get(i),
argsDigests.get(i));
+ CallResult pending = CallResult.pending(functionIds.get(i));
actionState.addCallResult(pending);
recoveryCallResults.add(pending);
}
@@ -1237,21 +1202,13 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
* call index.
*/
public void finalizeCurrentCall(
- String functionId,
- String argsDigest,
- byte[] resultPayload,
- byte[] exceptionPayload) {
- finalizeCallAt(
- currentCallIndex, functionId, argsDigest, resultPayload,
exceptionPayload);
+ String functionId, byte[] resultPayload, byte[]
exceptionPayload) {
+ finalizeCallAt(currentCallIndex, functionId, resultPayload,
exceptionPayload);
currentCallIndex++;
}
public void finalizeCallAt(
- int index,
- String functionId,
- String argsDigest,
- byte[] resultPayload,
- byte[] exceptionPayload) {
+ int index, String functionId, byte[] resultPayload, byte[]
exceptionPayload) {
CallResult current = getCallResultAt(index);
if (current == null) {
throw new IllegalStateException(
@@ -1259,11 +1216,11 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
"Cannot finalize call at index %s because no
persisted slot exists",
index));
}
- if (!current.matches(functionId, argsDigest)) {
+ if (!current.matches(functionId)) {
throw new IllegalStateException(
String.format(
- "Cannot finalize call at index %s because the
persisted slot does not match functionId=%s, argsDigest=%s",
- index, functionId, argsDigest));
+ "Cannot finalize call at index %s because the
persisted slot does not match functionId=%s",
+ index, functionId));
}
if (!current.isPending()) {
throw new IllegalStateException(
@@ -1272,17 +1229,15 @@ public class RunnerContextImpl implements
RunnerContext, ExecutionReporter {
index));
}
- CallResult terminal =
- new CallResult(functionId, argsDigest, resultPayload,
exceptionPayload);
+ CallResult terminal = new CallResult(functionId, resultPayload,
exceptionPayload);
actionState.replaceCallResult(index, terminal);
recoveryCallResults.set(index, terminal);
persistActionState();
LOG.debug(
- "Finalized and persisted CallResult at index {}:
functionId={}, argsDigest={}",
+ "Finalized and persisted CallResult at index {}:
functionId={}",
index,
- functionId,
- argsDigest);
+ functionId);
}
public void advanceCallIndexBy(int count) {
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
index 7f4badb14..21c2e7a37 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
@@ -203,9 +203,8 @@ public class ActionStateSerdeTest {
ActionState originalState = new ActionState(inputEvent);
// Add call results
- CallResult result1 = new CallResult("module.func1", "digest1",
"result1".getBytes());
- CallResult result2 =
- new CallResult("module.func2", "digest2", null,
"exception".getBytes());
+ CallResult result1 = new CallResult("module.func1",
"result1".getBytes());
+ CallResult result2 = new CallResult("module.func2", null,
"exception".getBytes());
originalState.addCallResult(result1);
originalState.addCallResult(result2);
@@ -218,14 +217,12 @@ public class ActionStateSerdeTest {
CallResult deserializedResult1 = deserializedState.getCallResult(0);
assertEquals("module.func1", deserializedResult1.getFunctionId());
- assertEquals("digest1", deserializedResult1.getArgsDigest());
assertArrayEquals("result1".getBytes(),
deserializedResult1.getResultPayload());
assertNull(deserializedResult1.getExceptionPayload());
assertTrue(deserializedResult1.isSuccess());
CallResult deserializedResult2 = deserializedState.getCallResult(1);
assertEquals("module.func2", deserializedResult2.getFunctionId());
- assertEquals("digest2", deserializedResult2.getArgsDigest());
assertNull(deserializedResult2.getResultPayload());
assertArrayEquals("exception".getBytes(),
deserializedResult2.getExceptionPayload());
assertTrue(deserializedResult2.isFailure());
@@ -235,7 +232,7 @@ public class ActionStateSerdeTest {
public void testActionStateWithPendingCallResult() throws Exception {
InputEvent inputEvent = new InputEvent("test input");
ActionState originalState = new ActionState(inputEvent);
- originalState.addCallResult(CallResult.pending("module.func",
"digest"));
+ originalState.addCallResult(CallResult.pending("module.func"));
byte[] serialized = ActionStateSerde.serialize(originalState);
ActionState deserializedState =
ActionStateSerde.deserialize(serialized);
@@ -282,8 +279,8 @@ public class ActionStateSerdeTest {
// Create in-progress ActionState with call results (simulating
partial execution)
InputEvent inputEvent = new InputEvent("test input");
List<CallResult> callResults = new ArrayList<>();
- callResults.add(new CallResult("func1", "hash1",
"result1".getBytes()));
- callResults.add(new CallResult("func2", "hash2",
"result2".getBytes()));
+ callResults.add(new CallResult("func1", "result1".getBytes()));
+ callResults.add(new CallResult("func2", "result2".getBytes()));
ActionState originalState =
new ActionState(inputEvent, null, null, null, callResults,
false);
@@ -295,8 +292,8 @@ public class ActionStateSerdeTest {
// Verify state
assertFalse(deserializedState.isCompleted());
assertEquals(2, deserializedState.getCallResultCount());
- assertTrue(deserializedState.getCallResult(0).matches("func1",
"hash1"));
- assertTrue(deserializedState.getCallResult(1).matches("func2",
"hash2"));
+ assertTrue(deserializedState.getCallResult(0).matches("func1"));
+ assertTrue(deserializedState.getCallResult(1).matches("func2"));
}
@Test
@@ -304,7 +301,7 @@ public class ActionStateSerdeTest {
// Test CallResult with null payloads
InputEvent inputEvent = new InputEvent("test");
ActionState originalState = new ActionState(inputEvent);
- originalState.addCallResult(new CallResult("func", "digest", null,
null));
+ originalState.addCallResult(new CallResult("func", null, null));
byte[] serialized = ActionStateSerde.serialize(originalState);
ActionState deserializedState =
ActionStateSerde.deserialize(serialized);
@@ -312,7 +309,6 @@ public class ActionStateSerdeTest {
assertEquals(1, deserializedState.getCallResultCount());
CallResult result = deserializedState.getCallResult(0);
assertEquals("func", result.getFunctionId());
- assertEquals("digest", result.getArgsDigest());
assertNull(result.getResultPayload());
assertNull(result.getExceptionPayload());
assertTrue(result.isSuccess());
@@ -335,7 +331,6 @@ public class ActionStateSerdeTest {
+ "\"callResults\":["
+ "{"
+ "\"functionId\":\"legacy.success\","
- + "\"argsDigest\":\"digest-success\","
+ "\"resultPayload\":\""
+ legacySuccessPayload
+ "\","
@@ -343,7 +338,6 @@ public class ActionStateSerdeTest {
+ "},"
+ "{"
+ "\"functionId\":\"legacy.failure\","
- + "\"argsDigest\":\"digest-failure\","
+ "\"resultPayload\":null,"
+ "\"exceptionPayload\":\""
+ legacyFailurePayload
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateTest.java
index 17296c1f9..dcb761091 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateTest.java
@@ -53,8 +53,8 @@ public class ActionStateTest {
List<org.apache.flink.agents.api.Event> outputEvents = new
ArrayList<>();
outputEvents.add(new OutputEvent("output"));
List<CallResult> callResults = new ArrayList<>();
- callResults.add(new CallResult("func1", "digest1",
"result1".getBytes()));
- callResults.add(new CallResult("func2", "digest2",
"result2".getBytes()));
+ callResults.add(new CallResult("func1", "result1".getBytes()));
+ callResults.add(new CallResult("func2", "result2".getBytes()));
boolean completed = true;
ActionState state =
@@ -78,8 +78,8 @@ public class ActionStateTest {
public void testAddCallResult() {
ActionState state = new ActionState(new InputEvent("test"));
- CallResult result1 = new CallResult("func1", "digest1",
"result1".getBytes());
- CallResult result2 = new CallResult("func2", "digest2",
"result2".getBytes());
+ CallResult result1 = new CallResult("func1", "result1".getBytes());
+ CallResult result2 = new CallResult("func2", "result2".getBytes());
state.addCallResult(result1);
assertEquals(1, state.getCallResultCount());
@@ -93,8 +93,8 @@ public class ActionStateTest {
@Test
public void testReplaceCallResult() {
ActionState state = new ActionState(new InputEvent("test"));
- CallResult original = new CallResult("func1", "digest1",
"result1".getBytes());
- CallResult replacement = CallResult.pending("func1", "digest1");
+ CallResult original = new CallResult("func1", "result1".getBytes());
+ CallResult replacement = CallResult.pending("func1");
state.addCallResult(original);
state.replaceCallResult(0, replacement);
@@ -112,7 +112,7 @@ public class ActionStateTest {
assertNull(state.getCallResult(0));
assertNull(state.getCallResult(100));
- state.addCallResult(new CallResult("func", "digest",
"result".getBytes()));
+ state.addCallResult(new CallResult("func", "result".getBytes()));
assertNull(state.getCallResult(1));
assertNotNull(state.getCallResult(0));
}
@@ -120,8 +120,8 @@ public class ActionStateTest {
@Test
public void testClearCallResults() {
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func1", "digest1",
"result1".getBytes()));
- state.addCallResult(new CallResult("func2", "digest2",
"result2".getBytes()));
+ state.addCallResult(new CallResult("func1", "result1".getBytes()));
+ state.addCallResult(new CallResult("func2", "result2".getBytes()));
assertEquals(2, state.getCallResultCount());
state.clearCallResults();
@@ -132,10 +132,10 @@ public class ActionStateTest {
@Test
public void testClearCallResultsFrom() {
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func0", "digest0",
"result0".getBytes()));
- state.addCallResult(new CallResult("func1", "digest1",
"result1".getBytes()));
- state.addCallResult(new CallResult("func2", "digest2",
"result2".getBytes()));
- state.addCallResult(new CallResult("func3", "digest3",
"result3".getBytes()));
+ state.addCallResult(new CallResult("func0", "result0".getBytes()));
+ state.addCallResult(new CallResult("func1", "result1".getBytes()));
+ state.addCallResult(new CallResult("func2", "result2".getBytes()));
+ state.addCallResult(new CallResult("func3", "result3".getBytes()));
assertEquals(4, state.getCallResultCount());
// Clear from index 2 onwards (keep func0, func1)
@@ -149,7 +149,7 @@ public class ActionStateTest {
@Test
public void testClearCallResultsFromInvalidIndex() {
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func", "digest",
"result".getBytes()));
+ state.addCallResult(new CallResult("func", "result".getBytes()));
// Negative index - should do nothing
state.clearCallResultsFrom(-1);
@@ -163,8 +163,8 @@ public class ActionStateTest {
@Test
public void testClearCallResultsFromZero() {
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func1", "digest1",
"result1".getBytes()));
- state.addCallResult(new CallResult("func2", "digest2",
"result2".getBytes()));
+ state.addCallResult(new CallResult("func1", "result1".getBytes()));
+ state.addCallResult(new CallResult("func2", "result2".getBytes()));
// Clear from index 0 - should clear all
state.clearCallResultsFrom(0);
@@ -174,8 +174,8 @@ public class ActionStateTest {
@Test
public void testMarkCompleted() {
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func1", "digest1",
"result1".getBytes()));
- state.addCallResult(new CallResult("func2", "digest2",
"result2".getBytes()));
+ state.addCallResult(new CallResult("func1", "result1".getBytes()));
+ state.addCallResult(new CallResult("func2", "result2".getBytes()));
assertFalse(state.isCompleted());
assertEquals(2, state.getCallResultCount());
@@ -190,10 +190,10 @@ public class ActionStateTest {
public void testEqualsWithCallResultsAndCompleted() {
InputEvent event = new InputEvent("test");
List<CallResult> callResults1 = new ArrayList<>();
- callResults1.add(new CallResult("func", "digest",
"result".getBytes()));
+ callResults1.add(new CallResult("func", "result".getBytes()));
List<CallResult> callResults2 = new ArrayList<>();
- callResults2.add(new CallResult("func", "digest",
"result".getBytes()));
+ callResults2.add(new CallResult("func", "result".getBytes()));
ActionState state1 = new ActionState(event, null, null, null,
callResults1, true);
ActionState state2 = new ActionState(event, null, null, null,
callResults2, true);
@@ -207,7 +207,7 @@ public class ActionStateTest {
public void testHashCodeWithCallResultsAndCompleted() {
InputEvent event = new InputEvent("test");
List<CallResult> callResults = new ArrayList<>();
- callResults.add(new CallResult("func", "digest", "result".getBytes()));
+ callResults.add(new CallResult("func", "result".getBytes()));
ActionState state1 = new ActionState(event, null, null, null,
callResults, true);
ActionState state2 =
@@ -219,7 +219,7 @@ public class ActionStateTest {
@Test
public void testToStringIncludesNewFields() {
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func", "digest",
"result".getBytes()));
+ state.addCallResult(new CallResult("func", "result".getBytes()));
state.markCompleted();
String str = state.toString();
@@ -253,13 +253,13 @@ public class ActionStateTest {
assertEquals(0, state.getCallResultCount());
// 2. First code block completes
- CallResult result1 = new CallResult("llm.call", "hash1",
"response1".getBytes());
+ CallResult result1 = new CallResult("llm.call",
"response1".getBytes());
state.addCallResult(result1);
assertEquals(1, state.getCallResultCount());
assertFalse(state.isCompleted());
// 3. Second code block completes
- CallResult result2 = new CallResult("db.query", "hash2",
"data".getBytes());
+ CallResult result2 = new CallResult("db.query", "data".getBytes());
state.addCallResult(result2);
assertEquals(2, state.getCallResultCount());
@@ -282,19 +282,19 @@ public class ActionStateTest {
// State from before failure (with 2 completed code blocks)
ActionState recoveredState = new ActionState(new InputEvent("test"));
- recoveredState.addCallResult(new CallResult("func1", "digest1",
"result1".getBytes()));
- recoveredState.addCallResult(new CallResult("func2", "digest2",
"result2".getBytes()));
+ recoveredState.addCallResult(new CallResult("func1",
"result1".getBytes()));
+ recoveredState.addCallResult(new CallResult("func2",
"result2".getBytes()));
// Check if action is completed
assertFalse(recoveredState.isCompleted());
// During re-execution, check if call result matches
CallResult result0 = recoveredState.getCallResult(0);
- assertTrue(result0.matches("func1", "digest1"));
+ assertTrue(result0.matches("func1"));
assertTrue(result0.isSuccess());
CallResult result1 = recoveredState.getCallResult(1);
- assertTrue(result1.matches("func2", "digest2"));
+ assertTrue(result1.matches("func2"));
// Third call is new (not in results)
assertNull(recoveredState.getCallResult(2));
@@ -304,17 +304,17 @@ public class ActionStateTest {
public void testNonDeterministicRecovery() {
// Simulate detection of non-deterministic call order
ActionState state = new ActionState(new InputEvent("test"));
- state.addCallResult(new CallResult("func1", "digest1",
"result1".getBytes()));
- state.addCallResult(new CallResult("func2", "digest2",
"result2".getBytes()));
- state.addCallResult(new CallResult("func3", "digest3",
"result3".getBytes()));
+ state.addCallResult(new CallResult("func1", "result1".getBytes()));
+ state.addCallResult(new CallResult("func2", "result2".getBytes()));
+ state.addCallResult(new CallResult("func3", "result3".getBytes()));
// During recovery, call 1 matches
CallResult result0 = state.getCallResult(0);
- assertTrue(result0.matches("func1", "digest1"));
+ assertTrue(result0.matches("func1"));
// Call 2 doesn't match (different function called)
CallResult result1 = state.getCallResult(1);
- assertFalse(result1.matches("different_func", "digest2"));
+ assertFalse(result1.matches("different_func"));
// Clear results from index 1 onwards
state.clearCallResultsFrom(1);
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/CallResultTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/CallResultTest.java
index 8e3a1af15..a6faa0205 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/CallResultTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/CallResultTest.java
@@ -27,13 +27,11 @@ public class CallResultTest {
@Test
public void testSuccessfulCallResult() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] resultPayload = "result".getBytes();
- CallResult result = new CallResult(functionId, argsDigest,
resultPayload);
+ CallResult result = new CallResult(functionId, resultPayload);
assertEquals(functionId, result.getFunctionId());
- assertEquals(argsDigest, result.getArgsDigest());
assertArrayEquals(resultPayload, result.getResultPayload());
assertNull(result.getExceptionPayload());
assertTrue(result.isSuccess());
@@ -42,13 +40,11 @@ public class CallResultTest {
@Test
public void testFailedCallResult() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] exceptionPayload = "exception".getBytes();
- CallResult result = new CallResult(functionId, argsDigest, null,
exceptionPayload);
+ CallResult result = new CallResult(functionId, null, exceptionPayload);
assertEquals(functionId, result.getFunctionId());
- assertEquals(argsDigest, result.getArgsDigest());
assertNull(result.getResultPayload());
assertArrayEquals(exceptionPayload, result.getExceptionPayload());
assertTrue(result.isFailure());
@@ -57,14 +53,12 @@ public class CallResultTest {
@Test
public void testFullConstructor() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] resultPayload = "result".getBytes();
byte[] exceptionPayload = null;
- CallResult result = new CallResult(functionId, argsDigest,
resultPayload, exceptionPayload);
+ CallResult result = new CallResult(functionId, resultPayload,
exceptionPayload);
assertEquals(functionId, result.getFunctionId());
- assertEquals(argsDigest, result.getArgsDigest());
assertArrayEquals(resultPayload, result.getResultPayload());
assertNull(result.getExceptionPayload());
assertTrue(result.isSuccess());
@@ -72,7 +66,7 @@ public class CallResultTest {
@Test
public void testPendingCallResult() {
- CallResult result = CallResult.pending("my_module.my_function",
"abc123");
+ CallResult result = CallResult.pending("my_module.my_function");
assertNull(result.getResultPayload());
assertNull(result.getExceptionPayload());
@@ -82,11 +76,9 @@ public class CallResultTest {
@Test
public void testLegacyStatusInference() {
CallResult success =
- CallResult.ofNullStatus(
- "my_module.my_function", "abc123",
"result".getBytes(), null);
+ CallResult.ofNullStatus("my_module.my_function",
"result".getBytes(), null);
CallResult failure =
- CallResult.ofNullStatus(
- "my_module.my_function", "abc123", null,
"exception".getBytes());
+ CallResult.ofNullStatus("my_module.my_function", null,
"exception".getBytes());
assertTrue(success.isSuccess());
assertTrue(failure.isFailure());
@@ -95,35 +87,33 @@ public class CallResultTest {
@Test
public void testMatches() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] resultPayload = "result".getBytes();
- CallResult result = new CallResult(functionId, argsDigest,
resultPayload);
+ CallResult result = new CallResult(functionId, resultPayload);
- assertTrue(result.matches(functionId, argsDigest));
- assertFalse(result.matches("other_function", argsDigest));
- assertFalse(result.matches(functionId, "other_digest"));
- assertFalse(result.matches("other_function", "other_digest"));
+ assertTrue(result.matches(functionId));
+ assertFalse(result.matches("other_function"));
+ assertTrue(result.matches(functionId));
+ assertFalse(result.matches("other_function"));
}
@Test
public void testMatchesWithNullValues() {
CallResult result = new CallResult();
- assertTrue(result.matches(null, null));
- assertFalse(result.matches("function", null));
- assertFalse(result.matches(null, "digest"));
+ assertTrue(result.matches(null));
+ assertFalse(result.matches("function"));
+ assertTrue(result.matches(null));
}
@Test
public void testEquals() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] resultPayload = "result".getBytes();
- CallResult result1 = new CallResult(functionId, argsDigest,
resultPayload);
- CallResult result2 = new CallResult(functionId, argsDigest,
resultPayload);
- CallResult result3 = new CallResult("other", argsDigest,
resultPayload);
+ CallResult result1 = new CallResult(functionId, resultPayload);
+ CallResult result2 = new CallResult(functionId, resultPayload);
+ CallResult result3 = new CallResult("other", resultPayload);
assertEquals(result1, result2);
assertNotEquals(result1, result3);
@@ -135,11 +125,10 @@ public class CallResultTest {
@Test
public void testHashCode() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] resultPayload = "result".getBytes();
- CallResult result1 = new CallResult(functionId, argsDigest,
resultPayload);
- CallResult result2 = new CallResult(functionId, argsDigest,
resultPayload);
+ CallResult result1 = new CallResult(functionId, resultPayload);
+ CallResult result2 = new CallResult(functionId, resultPayload);
assertEquals(result1.hashCode(), result2.hashCode());
}
@@ -147,20 +136,18 @@ public class CallResultTest {
@Test
public void testToString() {
String functionId = "my_module.my_function";
- String argsDigest = "abc123";
byte[] resultPayload = "result".getBytes();
- CallResult result = new CallResult(functionId, argsDigest,
resultPayload);
+ CallResult result = new CallResult(functionId, resultPayload);
String str = result.toString();
assertTrue(str.contains(functionId));
- assertTrue(str.contains(argsDigest));
assertTrue(str.contains("bytes"));
}
@Test
public void testToStringWithNullPayloads() {
- CallResult result = new CallResult("func", "digest", null, null);
+ CallResult result = new CallResult("func", null, null);
String str = result.toString();
assertTrue(str.contains("null"));
@@ -172,7 +159,6 @@ public class CallResultTest {
CallResult result = new CallResult();
assertNull(result.getFunctionId());
- assertNull(result.getArgsDigest());
assertNull(result.getResultPayload());
assertNull(result.getExceptionPayload());
assertTrue(result.isSuccess()); // exceptionPayload is null
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/DurableExecutionContextTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/DurableExecutionContextTest.java
index bdcd065d9..3e1b5fea2 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/DurableExecutionContextTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/DurableExecutionContextTest.java
@@ -90,9 +90,9 @@ class DurableExecutionContextTest {
@Test
void testInitialization() {
actionState.addCallResult(
- new CallResult("funcA", "digestA",
"resultA".getBytes(StandardCharsets.UTF_8)));
+ new CallResult("funcA",
"resultA".getBytes(StandardCharsets.UTF_8)));
actionState.addCallResult(
- new CallResult("funcB", "digestB",
"resultB".getBytes(StandardCharsets.UTF_8)));
+ new CallResult("funcB",
"resultB".getBytes(StandardCharsets.UTF_8)));
RunnerContextImpl.DurableExecutionContext context = createContext();
@@ -103,11 +103,11 @@ class DurableExecutionContextTest {
@Test
void testMatchNextOrClearSubsequentCallResultHit() {
byte[] expectedResult =
"cached_result".getBytes(StandardCharsets.UTF_8);
- actionState.addCallResult(new CallResult("funcA", "digestA",
expectedResult));
+ actionState.addCallResult(new CallResult("funcA", expectedResult));
RunnerContextImpl.DurableExecutionContext context = createContext();
- Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA", "digestA");
+ Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA");
assertNotNull(result);
assertEquals(3, result.length);
@@ -121,7 +121,7 @@ class DurableExecutionContextTest {
void testMatchNextOrClearSubsequentCallResultMiss() {
RunnerContextImpl.DurableExecutionContext context = createContext();
- Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA", "digestA");
+ Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA");
assertNull(result);
assertEquals(0, context.getCurrentCallIndex());
@@ -129,11 +129,11 @@ class DurableExecutionContextTest {
@Test
void testMatchNextOrClearSubsequentCallResultPendingTreatedAsMiss() {
- actionState.addCallResult(CallResult.pending("funcA", "digestA"));
+ actionState.addCallResult(CallResult.pending("funcA"));
RunnerContextImpl.DurableExecutionContext context = createContext();
- Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA", "digestA");
+ Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA");
assertNull(result);
assertEquals(0, context.getCurrentCallIndex());
@@ -142,13 +142,13 @@ class DurableExecutionContextTest {
@Test
void testMatchNextOrClearSubsequentCallResultMismatch() {
- actionState.addCallResult(new CallResult("funcA", "digestA",
"result".getBytes()));
- actionState.addCallResult(new CallResult("funcB", "digestB",
"result".getBytes()));
+ actionState.addCallResult(new CallResult("funcA",
"result".getBytes()));
+ actionState.addCallResult(new CallResult("funcB",
"result".getBytes()));
RunnerContextImpl.DurableExecutionContext context = createContext();
// Call with mismatched functionId - should clear subsequent results
and return null
- Object[] result =
context.matchNextOrClearSubsequentCallResult("funcX", "digestX");
+ Object[] result =
context.matchNextOrClearSubsequentCallResult("funcX");
assertNull(result);
// ActionState should have results cleared from index 0
@@ -162,7 +162,7 @@ class DurableExecutionContextTest {
RunnerContextImpl.DurableExecutionContext context = createContext();
byte[] resultPayload =
"success_result".getBytes(StandardCharsets.UTF_8);
- context.recordCallCompletion("funcA", "digestA", resultPayload, null);
+ context.recordCallCompletion("funcA", resultPayload, null);
assertEquals(1, context.getCurrentCallIndex());
assertEquals(1, actionState.getCallResults().size());
@@ -177,7 +177,7 @@ class DurableExecutionContextTest {
RunnerContextImpl.DurableExecutionContext context = createContext();
byte[] exceptionPayload =
"exception_data".getBytes(StandardCharsets.UTF_8);
- context.recordCallCompletion("funcA", "digestA", null,
exceptionPayload);
+ context.recordCallCompletion("funcA", null, exceptionPayload);
assertEquals(1, context.getCurrentCallIndex());
CallResult recorded = actionState.getCallResults().get(0);
@@ -190,36 +190,36 @@ class DurableExecutionContextTest {
void testMultipleCallResultRecovery() {
byte[] result1 = "result1".getBytes(StandardCharsets.UTF_8);
byte[] result2 = "result2".getBytes(StandardCharsets.UTF_8);
- actionState.addCallResult(new CallResult("func1", "digest1", result1));
- actionState.addCallResult(new CallResult("func2", "digest2", result2));
+ actionState.addCallResult(new CallResult("func1", result1));
+ actionState.addCallResult(new CallResult("func2", result2));
RunnerContextImpl.DurableExecutionContext context = createContext();
// First call should hit
- Object[] hit1 = context.matchNextOrClearSubsequentCallResult("func1",
"digest1");
+ Object[] hit1 = context.matchNextOrClearSubsequentCallResult("func1");
assertNotNull(hit1);
assertTrue((Boolean) hit1[0]);
assertArrayEquals(result1, (byte[]) hit1[1]);
// Second call should hit
- Object[] hit2 = context.matchNextOrClearSubsequentCallResult("func2",
"digest2");
+ Object[] hit2 = context.matchNextOrClearSubsequentCallResult("func2");
assertNotNull(hit2);
assertTrue((Boolean) hit2[0]);
assertArrayEquals(result2, (byte[]) hit2[1]);
// Third call should miss (no more results)
- Object[] miss = context.matchNextOrClearSubsequentCallResult("func3",
"digest3");
+ Object[] miss = context.matchNextOrClearSubsequentCallResult("func3");
assertNull(miss);
}
@Test
void testRecoveryWithExceptionPayload() {
byte[] exceptionPayload =
"exception_data".getBytes(StandardCharsets.UTF_8);
- actionState.addCallResult(new CallResult("funcA", "digestA", null,
exceptionPayload));
+ actionState.addCallResult(new CallResult("funcA", null,
exceptionPayload));
RunnerContextImpl.DurableExecutionContext context = createContext();
- Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA", "digestA");
+ Object[] result =
context.matchNextOrClearSubsequentCallResult("funcA");
assertNotNull(result);
assertTrue((Boolean) result[0]); // isHit
@@ -232,9 +232,9 @@ class DurableExecutionContextTest {
RunnerContextImpl.DurableExecutionContext context = createContext();
// Record multiple completions
- context.recordCallCompletion("func1", "digest1", "result1".getBytes(),
null);
- context.recordCallCompletion("func2", "digest2", "result2".getBytes(),
null);
- context.recordCallCompletion("func3", "digest3", "result3".getBytes(),
null);
+ context.recordCallCompletion("func1", "result1".getBytes(), null);
+ context.recordCallCompletion("func2", "result2".getBytes(), null);
+ context.recordCallCompletion("func3", "result3".getBytes(), null);
// Each call should trigger persistence
assertEquals(3, persistCallCount.get());
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 35f9ed08d..2c345c96c 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
@@ -128,7 +128,7 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
actionState.addCallResult(
- new CallResult("recon-async", "",
OBJECT_MAPPER.writeValueAsBytes("cached")));
+ new CallResult("recon-async",
OBJECT_MAPPER.writeValueAsBytes("cached")));
JavaRunnerContextImpl context = createContext(actionState, executor);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
@@ -150,7 +150,7 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
void testDurableExecuteAsyncReconcilableReplayNullSuccess() throws
Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
- actionState.addCallResult(new CallResult("recon-async", "", null,
null));
+ actionState.addCallResult(new CallResult("recon-async", null, null));
JavaRunnerContextImpl context = createContext(actionState, executor);
TestReconcilableCallable<Void> callable =
new TestReconcilableCallable<>(
@@ -171,7 +171,7 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
void testDurableExecuteAsyncReconcilableReconcileSuccess() throws
Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("recon-async", ""));
+ actionState.addCallResult(CallResult.pending("recon-async"));
JavaRunnerContextImpl context = createContext(actionState, executor);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
@@ -196,7 +196,7 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
void
testDurableExecuteAsyncReconcilableReconcileExceptionPersistsFailure() throws
Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("recon-async", ""));
+ actionState.addCallResult(CallResult.pending("recon-async"));
JavaRunnerContextImpl context = createContext(actionState, executor);
executor.setBeforeExecute(
() -> {
@@ -236,7 +236,7 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
void testDurableExecuteAsyncCompletionOnlyReExecutesPendingSlot() throws
Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("tool-call", ""));
+ actionState.addCallResult(CallResult.pending("tool-call"));
JavaRunnerContextImpl context = createContext(actionState, executor);
TestDurableCallable<String> callable =
new TestDurableCallable<>("tool-call", String.class, () ->
"recovered");
@@ -485,7 +485,7 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
void testGatherReconcilesPendingSlot() throws Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("batch-1", ""));
+ actionState.addCallResult(CallResult.pending("batch-1"));
JavaRunnerContextImpl context = createContext(actionState, executor);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
@@ -510,10 +510,10 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
ActionState actionState = new ActionState(null);
actionState.addCallResult(
- new CallResult("batch-1", "",
OBJECT_MAPPER.writeValueAsBytes("cached-one")));
+ new CallResult("batch-1",
OBJECT_MAPPER.writeValueAsBytes("cached-one")));
actionState.addCallResult(
- new CallResult("batch-2", "",
OBJECT_MAPPER.writeValueAsBytes("cached-two")));
- actionState.addCallResult(CallResult.pending("batch-3", ""));
+ new CallResult("batch-2",
OBJECT_MAPPER.writeValueAsBytes("cached-two")));
+ actionState.addCallResult(CallResult.pending("batch-3"));
JavaRunnerContextImpl context = createContext(actionState, executor);
TestDurableCallable<String> first =
new TestDurableCallable<>(
@@ -547,7 +547,6 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
actionState.addCallResult(
new CallResult(
"batch-1",
- "",
null,
OBJECT_MAPPER.writeValueAsBytes(
RunnerContextImpl.DurableExecutionException.fromException(
@@ -575,7 +574,6 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
actionState.addCallResult(
new CallResult(
"batch-1",
- "",
"not-valid-json".getBytes(java.nio.charset.StandardCharsets.UTF_8),
null));
JavaRunnerContextImpl context = createContext(actionState, executor);
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 2cad205e2..e17dc12c4 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
@@ -151,7 +151,7 @@ class RunnerContextImplDurableExecuteTest {
@Test
void
testDurableExecuteCompletionOnlyReExecutesPendingSlotDoesNotPersistInterruption()
{
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("tool-call", ""));
+ actionState.addCallResult(CallResult.pending("tool-call"));
RunnerContextImpl context = createContext(actionState);
TestDurableCallable<String> callable =
new TestDurableCallable<>(
@@ -290,7 +290,7 @@ class RunnerContextImplDurableExecuteTest {
void testDurableExecuteReconcilableReplaySuccess() throws Exception {
ActionState actionState = new ActionState(null);
actionState.addCallResult(
- new CallResult("recon-call", "",
OBJECT_MAPPER.writeValueAsBytes("cached")));
+ new CallResult("recon-call",
OBJECT_MAPPER.writeValueAsBytes("cached")));
RunnerContextImpl context = createContext(actionState);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
@@ -311,7 +311,7 @@ class RunnerContextImplDurableExecuteTest {
@Test
void testDurableExecuteReconcilableReplayNullSuccess() throws Exception {
ActionState actionState = new ActionState(null);
- actionState.addCallResult(new CallResult("recon-call", "", null,
null));
+ actionState.addCallResult(new CallResult("recon-call", null, null));
RunnerContextImpl context = createContext(actionState);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
@@ -334,7 +334,6 @@ class RunnerContextImplDurableExecuteTest {
actionState.addCallResult(
new CallResult(
"recon-call",
- "",
null,
OBJECT_MAPPER.writeValueAsBytes(
RunnerContextImpl.DurableExecutionException.fromException(
@@ -360,7 +359,7 @@ class RunnerContextImplDurableExecuteTest {
@Test
void testDurableExecuteReconcilableReconcileSuccess() throws Exception {
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("recon-call", ""));
+ actionState.addCallResult(CallResult.pending("recon-call"));
RunnerContextImpl context = createContext(actionState);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
@@ -384,7 +383,7 @@ class RunnerContextImplDurableExecuteTest {
@Test
void testDurableExecuteReconcilableReconcileExceptionPersistsFailure()
throws Exception {
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("recon-call", ""));
+ actionState.addCallResult(CallResult.pending("recon-call"));
RunnerContextImpl context = createContext(actionState);
IllegalStateException failure = new IllegalStateException("reconcile
unavailable");
TestReconcilableCallable<String> callable =
@@ -412,7 +411,7 @@ class RunnerContextImplDurableExecuteTest {
@Test
void testDurableExecuteCompletionOnlyReExecutesPendingSlot() throws
Exception {
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("tool-call", ""));
+ actionState.addCallResult(CallResult.pending("tool-call"));
RunnerContextImpl context = createContext(actionState);
TestDurableCallable<String> callable =
new TestDurableCallable<>("tool-call", String.class, () ->
"recovered");
@@ -433,8 +432,8 @@ class RunnerContextImplDurableExecuteTest {
void
testDurableExecuteCompletionOnlyReExecutesPendingSlotAfterBatchReservation()
throws Exception {
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("tool-call", ""));
- actionState.addCallResult(CallResult.pending("tool-call", ""));
+ actionState.addCallResult(CallResult.pending("tool-call"));
+ actionState.addCallResult(CallResult.pending("tool-call"));
RunnerContextImpl context = createContext(actionState);
TestDurableCallable<String> first =
new TestDurableCallable<>("tool-call", String.class, () ->
"one");
@@ -466,7 +465,7 @@ class RunnerContextImplDurableExecuteTest {
@Test
void testDurableExecuteReconcilableMismatchStartsNewCall() throws
Exception {
ActionState actionState = new ActionState(null);
- actionState.addCallResult(CallResult.pending("stale-call", ""));
+ actionState.addCallResult(CallResult.pending("stale-call"));
RunnerContextImpl context = createContext(actionState);
TestReconcilableCallable<String> callable =
new TestReconcilableCallable<>(
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 918356de2..617af4c0b 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
@@ -3637,7 +3637,7 @@ public class ActionExecutionOperatorTest {
input,
agentPlan,
"durableReconcilableAction",
-
actionStateWithCallResults(CallResult.pending("reconcilable-call", "")));
+
actionStateWithCallResults(CallResult.pending("reconcilable-call")));
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
@@ -3682,7 +3682,7 @@ public class ActionExecutionOperatorTest {
input,
agentPlan,
"durableReconcilableAction",
-
actionStateWithCallResults(CallResult.pending("reconcilable-call", "")));
+
actionStateWithCallResults(CallResult.pending("reconcilable-call")));
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
@@ -3725,7 +3725,7 @@ public class ActionExecutionOperatorTest {
input,
agentPlan,
"durableReconcilableAction",
- actionStateWithCallResults(CallResult.pending("stale-call",
"")));
+ actionStateWithCallResults(CallResult.pending("stale-call")));
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
@@ -3771,9 +3771,8 @@ public class ActionExecutionOperatorTest {
agentPlan,
"durableMixedRecoveryAction",
actionStateWithCallResults(
- new CallResult(
- "mixed-legacy-call", "",
OBJECT_MAPPER.writeValueAsBytes(11L)),
- CallResult.pending("mixed-reconcilable-call", "")));
+ new CallResult("mixed-legacy-call",
OBJECT_MAPPER.writeValueAsBytes(11L)),
+ CallResult.pending("mixed-reconcilable-call")));
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
