You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "da-daken (via GitHub)" <gi...@apache.org> on 2026/07/24 06:08:07 UTC
[PR] Parallel tool [flink-agents]
da-daken opened a new pull request, #926:
URL: https://github.com/apache/flink-agents/pull/926
<!--
* Thank you very much for contributing to Flink Agents.
* Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc.
-->
<!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. -->
Linked issue: https://github.com/apache/flink-agents/issues/925
### Purpose of change
Add parallel execution capability for tool calls via a new public API (`RunnerContext.durableExecuteAllAsync()`), which submits a batch of tool calls concurrently to an async thread pool and returns results in the original order after all complete.
The execution is durable: the state of in-flight calls is persisted, so that upon recovery (e.g., from checkpoint) the system correctly restores and completes the pending calls, ensuring no duplicate or lost invocations.
### Tests
- Unit tests to verify the ordering and consistency of parallel execution results.
- Integration tests to validate correctness under concurrent scenarios.
- Fault-tolerance tests to ensure that after a simulated failure and recovery, the state of parallel executions is correctly restored and all results are eventually returned exactly once.
### API
- Introduce a new public API `RunnerContext.durableExecuteAllAsync()` for batch parallel execution of tool calls.
- Expose the parallel execution capability without breaking compatibility with the existing synchronous API.
### Documentation
<!-- Do not remove this section. Check the proper box only. -->
- [x] `doc-needed` <!-- Your PR changes impact docs -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-included` <!-- Your PR already contains the necessary documentation updates -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3665493624
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -636,6 +721,110 @@ def wrapped_func(*a: Any, **kw: Any) -> Any:
return wrapped_func
+ def _call_matches(
+ self, current: _PersistedCallResult, call: DurableCall, args_digest: str
+ ) -> bool:
+ return current.function_id == call.id and current.args_digest == args_digest
+
+ def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
+ if current.exception_payload is not None:
+ return Outcome.failure(cloudpickle.loads(current.exception_payload))
+ if current.result_payload is None:
+ return Outcome.success(None)
+ return Outcome.success(cloudpickle.loads(current.result_payload))
+
+ def _callable_for_durable_call(self, call: DurableCall) -> Callable[[], Any]:
+ kwargs = call.kwargs or {}
+ return partial(call.func, *call.args, **kwargs)
+
+ def _prepare_batch_execution(self, calls: list[DurableCall]) -> _BatchExecutionPlan:
+ args_digest = ""
Review Comment:
Yes, inconsistent IDs on the Python side would cause this issue.
Regarding `functionId` – I plan to address it in this PR as well. My thought is to make it exactly the same on both the Python and Java sides. What do you think? Alternatively, we can open a separate issue for further discussion if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3680736901
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -636,6 +721,110 @@ def wrapped_func(*a: Any, **kw: Any) -> Any:
return wrapped_func
+ def _call_matches(
+ self, current: _PersistedCallResult, call: DurableCall, args_digest: str
+ ) -> bool:
+ return current.function_id == call.id and current.args_digest == args_digest
+
+ def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
+ if current.exception_payload is not None:
+ return Outcome.failure(cloudpickle.loads(current.exception_payload))
+ if current.result_payload is None:
+ return Outcome.success(None)
+ return Outcome.success(cloudpickle.loads(current.result_payload))
+
+ def _callable_for_durable_call(self, call: DurableCall) -> Callable[[], Any]:
+ kwargs = call.kwargs or {}
+ return partial(call.func, *call.args, **kwargs)
+
+ def _prepare_batch_execution(self, calls: list[DurableCall]) -> _BatchExecutionPlan:
+ args_digest = ""
Review Comment:
Good catch — you're right that there’s still an upgrade cost here.
I agree we should split the broader semantics (“unique per-call id as recovery identity” / the same-args-different-result case) into a separate issue. For this PR I’ll revert the tool functionId back to the fixed "tool-call" on both Java and Python, so serial and batch stay consistent with the previous release and we don’t change recovery identity in this change.
Java keeps a fixed `"tool-call"` functionId with an empty argsDigest; Python uses `qualname + args hash`, aligned with serial recovery semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3697315162
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -412,8 +485,9 @@ def _try_get_cached_result(
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 = _compute_function_id(func)
- args_digest = _compute_args_digest(args, kwargs)
+ function_id, args_digest = _resolve_durable_identity(
Review Comment:
The new PENDING branch seems to open a recovery gap on Python, and I would welcome a sanity check on the routing. `matchNextOrClearSubsequentCallResult` returns `null` for a matching PENDING slot without clearing it or advancing `currentCallIndex` (`RunnerContextImpl.java:762-771`). Java pre-checks PENDING in `durableExecuteCompletionOnly` at `:297-300` and finalizes in place, so it never lands there. Python's `_try_get_cached_result` reads that `null` as a plain miss (`:494`), re-executes, and `recordCallCompletion` appends (`RunnerContextImpl.java:813`): a two-call batch replayed serially becomes `[PENDING(f0), PENDING(f1), SUCCEEDED(f0), SUCCEEDED(f1)]`, and each later failover re-runs both tools and appends two more.
It is new at this head: at `9169bb2` the batch reserved `"tool-call-<call_id>"`, which never matched the serial digest, so the mismatch branch cleared and re-ran cleanly. No config flip is needed either, since `len(executions)` dropping to 1 on an unresolvable tool routes a reserved batch to the serial path (`tool_call_action.py:69`, `:108-117`); flipping back to parallel caps the growth but leaves a stale tail. Java has the recovery test (`RunnerContextImplDurableExecuteTest.java:267-299`); would a Python counterpart make sense, given the parity ask in `AGENTS.md`? For the fix, the bridge could take Java's pre-check, or the matcher could clear the slot before returning the miss. Which fits the bridge contract better?
##########
docs/content/docs/operations/configuration.md:
##########
@@ -131,7 +131,10 @@ Here is the list of all built-in core configuration options.
| `max-retries` | 3 | int | Number of retries when using `ErrorHandlingStrategy.RETRY`. |
| `retry-wait-interval` | 1 | int | Base wait interval in seconds between retries when using `ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the connection name. |
| `chat.async` | true | boolean | Whether chat asynchronously for built-in chat action. |
-| `tool-call.async` | true | boolean | Whether process tool call for built-in tool call action. |
+| `tool-call.async` | true | boolean | Whether the built-in tool-call action runs each tool via durable async execution. |
+| `tool-call.parallel` | true | boolean | When `tool-call.async` is also true (JDK 21+), run multiple tool calls from one `ToolRequestEvent` as one parallel durable batch. Increases in-flight external calls; after failover, unfinished tools may be submitted again — side-effecting tools should be idempotent or provide a reconciler. Set to `false` for serial tool execution. |
+| `tool-call.num-async-threads` | os cpu count * 2 | int | Dedicated thread pool size for tool-call async / parallel batch execution. Separate from `num-async-threads` so a large tool batch does not exhaust the global async pool. |
+| `tool-call.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall timeout for one parallel tool-call batch. Non-positive disables it. On timeout, completed slots keep their outcome and unfinished slots fail. Timeout cancellation is best-effort; external side effects from unfinished tool calls may still complete, so side-effecting tools should be idempotent or provide a reconciler. |
Review Comment:
nit: two JDK wording points. The new row has no JDK caveat, though the option is inert below Java 21: the java11 fallback takes `timeout` and ignores it (`ContinuationActionExecutor.java:75-87`), and two of the five IT combos run Java 17. Conversely `:135` says `tool-call.parallel` needs JDK 21+, which does not hold for Python, whose batch runs on a `ThreadPoolExecutor` (`flink_runner_context.py:266`). Worth a clause on each?
##########
python/flink_agents/api/runner_context.py:
##########
@@ -24,12 +25,49 @@
from flink_agents.api.metric_group import MetricGroup
from flink_agents.api.resource import Resource, ResourceType
-__all__ = ["AsyncExecutionResult", "RunnerContext"]
+__all__ = ["AsyncExecutionResult", "DurableCall", "Outcome", "RunnerContext"]
if TYPE_CHECKING:
from flink_agents.api.memory_object import MemoryObject
+@dataclass(frozen=True)
+class DurableCall:
+ """A deterministic durable call entry for batch execution."""
+
+ id: str
Review Comment:
`id` is required here but the Python runtime never reads it: `_durable_identity` (`flink_runner_context.py:751-752`) recomputes identity from `call.func/args/kwargs` at `:755`, `:821`, `:843`. Java treats the id as the journal identity (`RunnerContextImpl.java:293`), so a stable `id` with varying arguments keeps recovery on Java and loses it on Python. `DurableCall` is new, so it is cheap to settle now. Should `id` stay on the dataclass at all?
Keeping it also pulls in the `plan` to `runtime` import (`tool_call_action.py:34`), the first production one in the repo, against the direction `AGENTS.md` states. Java declares the id locally (`ToolCallAction.java:45`). Where would you want that to live?
##########
python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py:
##########
@@ -128,27 +170,78 @@ def clearCallResultsFromCurrentIndexAndPersist(self) -> None:
self.operations.append("clear")
self.call_results = self.call_results[: self.current_call_index]
+ def reservePendingBatch(
+ self, function_ids: list[str], args_digests: list[str]
+ ) -> None:
+ self.operations.append(f"reserve:{len(function_ids)}")
+ for function_id, digest in zip(function_ids, args_digests, strict=True):
+ self.call_results.append(
+ _StoredCallResult(
+ function_id=function_id,
+ args_digest=digest,
+ status="PENDING",
Review Comment:
The fake's `matchNextOrClearSubsequentCallResult` (`:103-116`) has no `status == "PENDING"` branch, so a PENDING match still returns `[True, None, None]`, the pre-fix behaviour changed at `RunnerContextImpl.java:762-771`. A Python regression test for the batch to serial case would pass here while production appends. Would it help to teach the fake the new contract first, so such a test could fail?
##########
python/flink_agents/api/runner_context.py:
##########
@@ -203,6 +241,7 @@ def durable_execute(
func: Callable[[Any], Any],
*args: Any,
reconciler: Callable[[], Any] | None = None,
+ durable_id: str | None = None,
Review Comment:
`durable_id` is on both public abstract signatures (`:244`, `:306`) and threaded through `flink_runner_context.py`, but nothing passes it, and `test_tool_call_action.py:361` asserts it is not passed. Java has no counterpart. Two things that made me pause while it sits unused: `reconciler`'s docstring marks itself reserved and not forwarded to `func` (`:285-286`) where `durable_id`'s (`:287-290`) does not, so a tool declaring its own `durable_id` would have it swallowed, and every third-party `RunnerContext` inherits the abstract signature. Is a follow-up expected to use it?
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java:
##########
@@ -79,55 +103,160 @@ public static void processToolRequest(Event event, RunnerContext ctx) {
diagnosticError = e.getMessage();
}
- if (tool != null) {
- try {
- // Framework-owned injected args must win over model-provided values so hidden
- // context such as tenant ids cannot be spoofed by a tool call payload.
- mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
- ToolResponse response;
- final Tool toolRef = tool;
- final Map<String, Object> callArguments = mergedArguments;
- DurableCallable<ToolResponse> callable =
- new DurableCallable<>() {
- @Override
- public String getId() {
- return "tool-call";
- }
-
- @Override
- public Class<ToolResponse> getResultClass() {
- return ToolResponse.class;
- }
-
- @Override
- public ToolResponse call() throws Exception {
- return toolRef.call(new ToolParameters(callArguments));
- }
- };
- response =
- toolCallAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
- success.put(id, response.isSuccess());
- responses.put(id, response);
- if (!response.isSuccess() && response.getError() != null) {
- error.put(id, response.getError());
- }
- } catch (Exception e) {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s execute failed.", name)));
- error.put(id, e.getMessage());
- }
- } else {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s does not exist.", name)));
- error.put(id, diagnosticError != null ? diagnosticError : "Tool does not exist.");
+ if (tool == null) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s does not exist.", name)),
+ diagnosticError != null ? diagnosticError : "Tool does not exist.",
+ success,
+ error,
+ responses);
+ continue;
}
+
+ try {
+ // Framework-owned injected args must win over model-provided values so hidden
+ // context such as tenant ids cannot be spoofed by a tool call payload.
+ mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
+ } catch (Exception e) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s execute failed.", name)),
+ e.getMessage(),
+ success,
+ error,
+ responses);
+ continue;
+ }
+
+ final Tool toolRef = tool;
+ final Map<String, Object> callArguments = mergedArguments;
+ DurableCallable<ToolResponse> callable =
+ new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "tool-call-" + id;
+ }
+
+ @Override
+ public Class<ToolResponse> getResultClass() {
+ return ToolResponse.class;
+ }
+
+ @Override
+ public ToolResponse call() throws Exception {
+ return toolRef.call(new ToolParameters(callArguments));
+ }
+ };
+ executions.add(new ToolCallExecution(id, name, callable));
+ }
+ return executions;
+ }
+
+ private static void executeParallel(
+ List<ToolCallExecution> executions,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses) {
+ List<DurableCallable<ToolResponse>> callables = new ArrayList<>(executions.size());
+ for (ToolCallExecution execution : executions) {
+ callables.add(execution.callable);
+ }
+ try {
+ List<Outcome<ToolResponse>> outcomes = ctx.durableExecuteAllAsync(callables);
+ for (int i = 0; i < outcomes.size(); i++) {
+ recordOutcome(executions.get(i), outcomes.get(i), success, error, responses);
+ }
+ } catch (Exception e) {
Review Comment:
Partly. The Python catch-all is gone (`tool_call_action.py:145-156`), but the Java one at `ToolCallAction.java:168-177` is unchanged, so the two languages now do different things with the same failure. Was leaving Java for the follow-up the intent? I answered your two-layer question in the review comment, since it bears on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5467091891
@weiqingy I really appreciate your assistance on this PR. Excellent work!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3746344131
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
Review Comment:
I think this barrier can never be completed. `batchBarrier` is created here and handed to `setPendingBatchFuture`, but the only thing that completes it is `:235`, which runs after the `while (completed < batchSize)` loop finishes. The loop yields at `:231` and needs `executeAction` to resume it, and `executeAction:72-74` will not resume while `hasPendingAsync()` is true. `ContinuationContext.java:74-76` reports true for exactly as long as the barrier is pending. So after the first yield the action stops making progress. No thread blocks, the task just re-queues on the mailbox forever (`ActionExecutionOperator.java:424`).
With the default `tool-call.batch.timeout.ms = -1` (`AgentExecutionOptions.java:81-82`) there is no deadline to fall back on, since `getDeadlineNanos:295-299` returns `Long.MAX_VALUE`. Setting a timeout does not really rescue it either: the resume then lands straight in the timeout branch at `:200`, so the success path at `:235` is effectively unreachable.
This is on by default now, because `tool-call.parallelism` defaults to `availableProcessors()` (`:69-73`) and `ToolCallAction.java:69` sends any two-tool batch down the parallel path. CI seems to agree: `it-java [java-21]` is cancelled at the 40-minute timeout on all three Flink combos at head, green at `5c9a67e3a` and red from `b293e5cd6` on. In job `92813859324` the test that stalls is `ReActAgentTest`, which is already on main, so it is not just the new tests. The two `java-17` combos take the serial fallback and pass in about 10 minutes.
The old `CompletableFuture.allOf(...)` let the pool complete the barrier from outside, which is what made the gate work. Would going back to that shape, and using the window only to decide when to submit, be enough to fix it?
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
+ context.setPendingBatchFuture(batchBarrier, deadlineNanos);
+
+ while (completed < batchSize) {
+ if (System.nanoTime() >= deadlineNanos) {
+ TimeoutException exception =
+ new TimeoutException(
+ "Async durable batch execution timed out after " + timeout);
+ batchBarrier.cancel(true);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomesOnTimeout(slots, exception);
+ }
+
+ while (nextToSubmit < batchSize && countInFlight(slots, nextToSubmit) < parallelismLimit) {
+ int index = nextToSubmit++;
+ Callable<T> supplier = suppliers.get(index);
+ slots[index] =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return Outcome.success(supplier.call());
+ } catch (Exception e) {
+ return Outcome.failure(e);
+ }
+ }, asyncExecutor);
+ }
+
+ for (int i = 0; i < nextToSubmit; i++) {
+ if (!counted[i] && slots[i].isDone()) {
+ counted[i] = true;
+ completed++;
+ }
+ }
+
+ if (completed < batchSize) {
+ Continuation.yield(SCOPE);
+ }
+ }
+
+ batchBarrier.complete(null);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomes(Arrays.asList(slots));
+ }
+
+ private static <T> int countInFlight(
+ CompletableFuture<Outcome<T>>[] slots, int submittedCount) {
+ int inFlight = 0;
+ for (int i = 0; i < submittedCount; i++) {
+ if (!slots[i].isDone()) {
+ inFlight++;
+ }
+ }
+ return inFlight;
+ }
+
+
+ /**
+ * Collects per-slot outcomes after the batch barrier completes normally.
+ *
+ * <p>Each supplier already wraps success and failure into an {@link Outcome}, so {@code join()}
+ * returns that outcome rather than throwing for ordinary tool exceptions.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomes(
+ List<CompletableFuture<Outcome<T>>> futures) {
+ List<Outcome<T>> results = new ArrayList<>(futures.size());
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ results.add(future.join());
+ }
+ return results;
+ }
+
+ /**
+ * Collects per-slot outcomes when the batch deadline elapses.
+ *
+ * <p>Completed slots keep their success or failure outcome. Only slots that are still running
+ * (or become cancelled) are finalized as timeout failures. {@code cancel(true)} is attempted
+ * only for unfinished futures; a future that completes between the check and cancel stays
+ * non-cancelled and is collected as a normal outcome.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomesOnTimeout(
+ CompletableFuture<Outcome<T>>[] futures, TimeoutException timeoutException) {
+ List<Outcome<T>> results = new ArrayList<>(futures.length);
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ if (future == null) {
+ results.add(Outcome.failure(timeoutException));
Review Comment:
When the deadline hits, slots the window never reached are still `null`, so each one becomes `Outcome.failure(timeoutException)` here. `JavaRunnerContextImpl.finalizeExecutedOutcomes:185-201` then persists them through `finalizeCallAt` as `Status.FAILED`. With `parallelism=2`, a 10-tool batch that times out records eight tools that never started as permanently failed, and they come back as cached failures on replay. Python does the same at `flink_runner_context.py:379-381`.
This is separate from the resume problem above, since it sits on the timeout path that a fixed barrier would still take. Before the sliding window every supplier was submitted up front, so at least every tool had started. It is also close to the open question about timeout and the reconciler, though a slot that never started feels like a different case. Would leaving those slots PENDING, so recovery can still run them, fit better than recording them as failed?
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
+ context.setPendingBatchFuture(batchBarrier, deadlineNanos);
+
+ while (completed < batchSize) {
+ if (System.nanoTime() >= deadlineNanos) {
+ TimeoutException exception =
+ new TimeoutException(
+ "Async durable batch execution timed out after " + timeout);
+ batchBarrier.cancel(true);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomesOnTimeout(slots, exception);
+ }
+
+ while (nextToSubmit < batchSize && countInFlight(slots, nextToSubmit) < parallelismLimit) {
+ int index = nextToSubmit++;
+ Callable<T> supplier = suppliers.get(index);
+ slots[index] =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return Outcome.success(supplier.call());
+ } catch (Exception e) {
+ return Outcome.failure(e);
+ }
+ }, asyncExecutor);
+ }
+
+ for (int i = 0; i < nextToSubmit; i++) {
+ if (!counted[i] && slots[i].isDone()) {
+ counted[i] = true;
+ completed++;
+ }
+ }
+
+ if (completed < batchSize) {
+ Continuation.yield(SCOPE);
+ }
+ }
+
+ batchBarrier.complete(null);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomes(Arrays.asList(slots));
+ }
+
+ private static <T> int countInFlight(
+ CompletableFuture<Outcome<T>>[] slots, int submittedCount) {
+ int inFlight = 0;
+ for (int i = 0; i < submittedCount; i++) {
+ if (!slots[i].isDone()) {
+ inFlight++;
+ }
+ }
+ return inFlight;
+ }
+
+
+ /**
+ * Collects per-slot outcomes after the batch barrier completes normally.
+ *
+ * <p>Each supplier already wraps success and failure into an {@link Outcome}, so {@code join()}
+ * returns that outcome rather than throwing for ordinary tool exceptions.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomes(
+ List<CompletableFuture<Outcome<T>>> futures) {
+ List<Outcome<T>> results = new ArrayList<>(futures.size());
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ results.add(future.join());
+ }
+ return results;
+ }
+
+ /**
+ * Collects per-slot outcomes when the batch deadline elapses.
+ *
+ * <p>Completed slots keep their success or failure outcome. Only slots that are still running
+ * (or become cancelled) are finalized as timeout failures. {@code cancel(true)} is attempted
+ * only for unfinished futures; a future that completes between the check and cancel stays
+ * non-cancelled and is collected as a normal outcome.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomesOnTimeout(
+ CompletableFuture<Outcome<T>>[] futures, TimeoutException timeoutException) {
+ List<Outcome<T>> results = new ArrayList<>(futures.length);
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ if (future == null) {
+ results.add(Outcome.failure(timeoutException));
+ continue;
+ }
+ if (!future.isDone()) {
+ future.cancel(true);
Review Comment:
`cancel(true)` does not interrupt anything on a `CompletableFuture`, since `mayInterruptIfRunning` is ignored there, so a tool that hangs keeps its thread until it returns on its own. Python has the same limit (`flink_runner_context.py:383`), so this looks like a shared gap rather than a Java one.
What changed this round is which pool that thread comes from. With the dedicated tool pool gone, `JavaRunnerContextImpl.executeOutcomeSuppliers:244` submits to the same `continuationExecutor` as chat and RAG, so one hung tool now costs every key on the subtask a thread. The javadoc at `AgentExecutionOptions.java:57-67` covers the sizing side of this. The part it does not cover is that after a batch timeout the thread never comes back. Is there a way to bound that, or is documenting it the right call for v1?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -477,7 +599,7 @@ protected <T> Optional<T> tryGetCachedResult(
} else if (resultPayload != null) {
return Optional.of(OBJECT_MAPPER.readValue(resultPayload, resultClass));
} else {
- return Optional.of(null);
+ return Optional.empty();
Review Comment:
Changing `Optional.of(null)` to `Optional.empty()` fixes a real NPE, but it also makes a cached `null` look exactly like a cache miss, and both callers only check `isPresent()`.
On the completion-only path (`:302-316`) that means the call runs a second time, and `recordCallCompletion:827-846` appends another `CallResult` and advances `currentCallIndex` again after the hit already advanced it at `:799`, so every later slot in the action shifts by one. On the reconcile path (`:649-659`) it throws `IllegalStateException` saying the slot is not terminal, when it is.
Any durable call that legitimately returns `null` lands here, since `serializeDurableResult` returns null (`:687-692`) and `new CallResult(fid, digest, null, null)` counts as SUCCEEDED (`actionstate/CallResult.java:93-101`). Tool calls are safe because `ToolCallAction` always returns a `ToolResponse`, so this is really about user code calling `durableExecute`. Same shape as the Python miss you just fixed. Would a separate hit flag, or a sentinel for the absent slot, be enough to tell the two apart?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -332,6 +352,23 @@ public static DurableExecutionException fromException(Exception e) {
}
public Exception toException() {
+ if (exceptionClass == null) {
+ return new RuntimeException(message);
+ }
+ try {
+ Class<?> clazz = Class.forName(exceptionClass);
+ if (Exception.class.isAssignableFrom(clazz)) {
+ @SuppressWarnings("unchecked")
+ Class<? extends Exception> exceptionClazz = (Class<? extends Exception>) clazz;
+ try {
+ return exceptionClazz.getConstructor(String.class).newInstance(message);
+ } catch (NoSuchMethodException ignored) {
+ return exceptionClazz.getConstructor().newInstance();
Review Comment:
This is the `toException()` work you mentioned.
The no-arg fallback here drops `message`, so the rebuilt exception has `getMessage() == null`, and `ToolCallAction.recordExecutionException:236` then writes a null into the `ToolResponseEvent` error map. The old wrapper always carried `exceptionClass + ": " + message`. Could the message be carried through on this path too?
Less certain, and I have not tested it: `Class.forName(exceptionClass)` at `:359` is the one-arg form, so it resolves the class with `RunnerContextImpl`'s own loader rather than the user-code loader that `JavaActionTask:67` installs. Framework types like `TimeoutException` resolve either way, so whether this bites probably depends on where the flink-agents jars sit.
##########
python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py:
##########
@@ -415,3 +594,420 @@ def collect_kwargs(**kwargs: Any) -> dict[str, Any]:
_close_runner_context(ctx)
assert result == {}
+
+
+def test_flink_runner_context_durable_execute_all_async_runs_calls_in_parallel() -> None:
+ j_runner_context = _FakeJavaRunnerContext()
+ config = AgentConfiguration(
+ {"tool-call.batch.timeout.ms": -1, "tool-call.parallelism": 3}
+ )
+ ctx = _create_runner_context(j_runner_context, config=config, executor_workers=3)
+ sleep_seconds = 0.2
+
+ def slow_call(value: str) -> str:
+ time.sleep(sleep_seconds)
+ return value
+
+ try:
+ start = time.perf_counter()
+ outcomes = _run_async(
+ ctx.durable_execute_all_async(
+ [
+ _durable_call(slow_call, "one"),
+ _durable_call(slow_call, "two"),
+ _durable_call(slow_call, "three"),
+ ]
+ )
+ )
+ elapsed = time.perf_counter() - start
+ finally:
+ _close_runner_context(ctx)
+
+ assert [outcome.value for outcome in outcomes] == ["one", "two", "three"]
+ assert elapsed < sleep_seconds * 2
Review Comment:
This assertion is failing at head: `ut-python [macos-latest] [java-17] [python-3.12]` reports `assert 0.40510524999990594 < (0.2 * 2)`. A 2x margin on a 200 ms sleep leaves roughly 200 ms for thread start-up on a shared runner, and the suite has gone red on a different Python version on each of the last three commits. Would a wider margin, or checking that the intervals overlap instead of total wall-clock, hold up better here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5214607704
Thanks @weiqingy @pltbkd for the new review — every point was helpful.
I’ve removed the extra id parameter on the Python side. Whether Java and Python should use the same recovery identity is worth discussing in a follow-up issue, as @pltbkd noted in the first point. I’ve also aligned Python recovery behavior with Java: when a PENDING slot is read, we re-execute and recover in that same slot — that was my oversight, thanks for catching it.
The durableExecuteAllAsync-related changes haven’t been pushed yet — sorry for the confusion. The catch-all semantics I mentioned earlier no longer apply. What I meant was that system/infrastructure exceptions should propagate so users can see them; in ToolCallAction, we can keep the same approach as the serial path — a system failure surfaces as tool-level failures for all affected tools.
On Layer A serialization failures: I’d like to correct my earlier wording — we should return a failed Outcome without persisting the slot as failed; the slot stays PENDING. On JDK < 21, we also don’t persist the slot; we throw instead, so on recovery we run call/reconcile, which matches the intent. When a system error happens before the slot is persisted, running call/reconcile on recovery is also what ToolCallAction expects.
Re @pltbkd’s review: `durableExecuteAllAsync `already has a real use case — execution timeout. From the framework’s perspective that’s a failure; we surface a timeout exception so callers can retry or compensate (which would land in another slot). That made the old toException() behavior — wrapping everything in RuntimeException — more painful, so I extended it to restore the original exception type after deserialization.
On the two thread pools question: I agree with a single shared pool + a cap on how many threads one record/batch can occupy, with batch parallelism = cores. A second pool would use extra memory without real benefit over enlarging the shared pool; semantically, a dedicated pool for durableExecuteAllAsync isn’t necessary — one pool is enough. Because parallel tool execution is on by default, it can contend with other keys; I’ll call that out clearly in the ConfigOption docs so users notice the behavior change.
On configuration: agreed — we can simplify to a single knob: parallelism = 1 (serial) vs parallelism > 1 (parallel).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3792074080
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
+ context.setPendingBatchFuture(batchBarrier, deadlineNanos);
+
+ while (completed < batchSize) {
+ if (System.nanoTime() >= deadlineNanos) {
+ TimeoutException exception =
+ new TimeoutException(
+ "Async durable batch execution timed out after " + timeout);
+ batchBarrier.cancel(true);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomesOnTimeout(slots, exception);
+ }
+
+ while (nextToSubmit < batchSize && countInFlight(slots, nextToSubmit) < parallelismLimit) {
+ int index = nextToSubmit++;
+ Callable<T> supplier = suppliers.get(index);
+ slots[index] =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return Outcome.success(supplier.call());
+ } catch (Exception e) {
+ return Outcome.failure(e);
+ }
+ }, asyncExecutor);
+ }
+
+ for (int i = 0; i < nextToSubmit; i++) {
+ if (!counted[i] && slots[i].isDone()) {
+ counted[i] = true;
+ completed++;
+ }
+ }
+
+ if (completed < batchSize) {
+ Continuation.yield(SCOPE);
+ }
+ }
+
+ batchBarrier.complete(null);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomes(Arrays.asList(slots));
+ }
+
+ private static <T> int countInFlight(
+ CompletableFuture<Outcome<T>>[] slots, int submittedCount) {
+ int inFlight = 0;
+ for (int i = 0; i < submittedCount; i++) {
+ if (!slots[i].isDone()) {
+ inFlight++;
+ }
+ }
+ return inFlight;
+ }
+
+
+ /**
+ * Collects per-slot outcomes after the batch barrier completes normally.
+ *
+ * <p>Each supplier already wraps success and failure into an {@link Outcome}, so {@code join()}
+ * returns that outcome rather than throwing for ordinary tool exceptions.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomes(
+ List<CompletableFuture<Outcome<T>>> futures) {
+ List<Outcome<T>> results = new ArrayList<>(futures.size());
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ results.add(future.join());
+ }
+ return results;
+ }
+
+ /**
+ * Collects per-slot outcomes when the batch deadline elapses.
+ *
+ * <p>Completed slots keep their success or failure outcome. Only slots that are still running
+ * (or become cancelled) are finalized as timeout failures. {@code cancel(true)} is attempted
+ * only for unfinished futures; a future that completes between the check and cancel stays
+ * non-cancelled and is collected as a normal outcome.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomesOnTimeout(
+ CompletableFuture<Outcome<T>>[] futures, TimeoutException timeoutException) {
+ List<Outcome<T>> results = new ArrayList<>(futures.length);
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ if (future == null) {
+ results.add(Outcome.failure(timeoutException));
+ continue;
+ }
+ if (!future.isDone()) {
+ future.cancel(true);
Review Comment:
Agreed to add the explanation in the V1 version first. I've already committed the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3661644118
##########
api/src/main/java/org/apache/flink/agents/api/agents/AgentExecutionOptions.java:
##########
@@ -42,9 +44,43 @@ public class AgentExecutionOptions {
public static final ConfigOption<Boolean> CHAT_ASYNC =
new ConfigOption<>("chat.async", Boolean.class, true);
+ /** Whether the built-in tool-call action runs each tool via durable async execution. */
public static final ConfigOption<Boolean> TOOL_CALL_ASYNC =
new ConfigOption<>("tool-call.async", Boolean.class, true);
+ /**
+ * Whether multiple tool calls from one {@code ToolRequestEvent} run as one parallel durable
+ * batch when {@link #TOOL_CALL_ASYNC} is also enabled (JDK 21+).
+ *
+ * <p>Default is {@code true}. A parallel batch raises the number of in-flight external calls;
+ * after failover, tools whose results were not yet persisted may be submitted again.
+ * Side-effecting tools should be idempotent or provide a {@code reconciler()}. Set to {@code
+ * false} to keep serial async or sync tool execution.
+ */
+ public static final ConfigOption<Boolean> TOOL_CALL_PARALLEL =
+ new ConfigOption<>("tool-call.parallel", Boolean.class, true);
+
+ /**
+ * Size of the dedicated thread pool used for tool-call async and parallel batch execution.
+ *
+ * <p>Separate from {@link #NUM_ASYNC_THREADS} so a large tool batch does not exhaust the global
+ * async pool.
+ */
+ public static final ConfigOption<Integer> TOOL_CALL_NUM_ASYNC_THREADS =
+ new ConfigOption<>(
+ "tool-call.num-async-threads",
+ Integer.class,
+ Runtime.getRuntime().availableProcessors() * 2);
+
+ /**
+ * Overall timeout for one parallel tool-call batch.
+ *
+ * <p>Non-positive values disable the timeout. When the deadline elapses, unfinished slots are
+ * failed; slots that already completed keep their success or failure outcome.
+ */
+ public static final ConfigOption<Duration> TOOL_CALL_BATCH_TIMEOUT =
+ new ConfigOption<>("tool-call.batch.timeout", Duration.class, Duration.ofMillis(-1));
Review Comment:
This is the repo's first `ConfigOption<Duration>`, and I do not think it can be set from any documented route.
`AgentConfiguration.get()` dispatches `isAssignableFrom`, `String`, `Integer`, `Long`, `Float`, `Double`, `Boolean`, `isEnum()`, then `throw new ClassCastException`. There is no `Duration` branch, and YAML values arrive as `String` or `Integer`. So `getConfig().get(TOOL_CALL_BATCH_TIMEOUT)` at `JavaRunnerContextImpl.java:253` throws for any value a user sets, after `:119` has already persisted N PENDING slots, and the catch-all at `ToolCallAction.java:171` then reports every tool in the batch failed.
The programmatic route clears `get()` but not plan serialization: `AgentPlan.writeObject` uses a bare `new ObjectMapper()` with no `JavaTimeModule`, and I reproduced `InvalidDefinitionException` on `java.time.Duration` against the pinned jackson-databind. On Python, `core_options.py:271` is `config_type=int`, so the documented `30s` raises `ValueError` mid-action.
CI stays green because `JavaRunnerContextImplDurableExecuteAsyncTest.java:324-327` sets the option in-process, where `isAssignableFrom` short-circuits. And `check_java_python_config_options_parity.py:45` adds `"java.time.Duration": int` to the type table, so the harness now certifies exactly this pair.
`ConfigurationUtils.convertValue` is already imported in `AgentConfiguration` for the enum branch and it handles `Duration`; retyping this option to `Long` millis would instead let the harness stay strict. Do you have a preference? Asking because the batch deadline is the only bound on a hung tool wedging the fan-in, and it currently ships both default-off and unsettable.
##########
runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java:
##########
@@ -240,12 +397,61 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) {
return supplier.get();
}
+ @Override
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ java.time.Duration timeout) {
+ executeAllAsyncCallCount++;
+ executeAllAsyncBatchSizes.add(suppliers.size());
+ if (useTimeoutCollection) {
+ return collectTimedOutOutcomes(suppliers);
+ }
+ List<Outcome<T>> outcomes = new java.util.ArrayList<>(suppliers.size());
+ for (Callable<T> supplier : suppliers) {
+ try {
+ outcomes.add(Outcome.success(supplier.call()));
+ } catch (Exception e) {
+ outcomes.add(Outcome.failure(e));
+ }
+ }
+ return outcomes;
+ }
+
+ private <T> List<Outcome<T>> collectTimedOutOutcomes(List<Callable<T>> suppliers) {
Review Comment:
`testToolCallBatchExecutionIsActuallyParallel` covers the real java21 path well, fallback branch included, which matters because surefire runs the exploded `target/classes` and so never loads the `META-INF/versions/21` classes. The deadline leg looks like the remaining gap, and this stub is what makes it look covered.
`collectTimedOutOutcomes` hard-codes "index 0 succeeds, everything else is a `TimeoutException`" and never reads the `timeout` argument. So in `testDurableExecuteAllAsyncTimeoutKeepsCompletedOutcomes` the 10 ms deadline set at `:324-327` and the `Thread.sleep(100)` in the second supplier are both inert, supplier 1 is never invoked, and its `getCallCount()` is never asserted. A deadline computed with the wrong sign, a `cancel` on the wrong future, or a barrier that never resolves would all still pass. What it does cover, a timed-out slot persisted FAILED with the cursor still advancing by 2, is worth keeping.
Now that the e2e path exists, would a batch that overruns a short `tool-call.batch.timeout` be the cheapest way to get `getDeadlineNanos` and `collectBatchOutcomesOnTimeout` genuinely exercised? That may have to wait on the `Duration` conversion on `AgentExecutionOptions.java:82`, since setting the option is what trips the plan serializer.
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java:
##########
@@ -79,55 +103,160 @@ public static void processToolRequest(Event event, RunnerContext ctx) {
diagnosticError = e.getMessage();
}
- if (tool != null) {
- try {
- // Framework-owned injected args must win over model-provided values so hidden
- // context such as tenant ids cannot be spoofed by a tool call payload.
- mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
- ToolResponse response;
- final Tool toolRef = tool;
- final Map<String, Object> callArguments = mergedArguments;
- DurableCallable<ToolResponse> callable =
- new DurableCallable<>() {
- @Override
- public String getId() {
- return "tool-call";
- }
-
- @Override
- public Class<ToolResponse> getResultClass() {
- return ToolResponse.class;
- }
-
- @Override
- public ToolResponse call() throws Exception {
- return toolRef.call(new ToolParameters(callArguments));
- }
- };
- response =
- toolCallAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
- success.put(id, response.isSuccess());
- responses.put(id, response);
- if (!response.isSuccess() && response.getError() != null) {
- error.put(id, response.getError());
- }
- } catch (Exception e) {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s execute failed.", name)));
- error.put(id, e.getMessage());
- }
- } else {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s does not exist.", name)));
- error.put(id, diagnosticError != null ? diagnosticError : "Tool does not exist.");
+ if (tool == null) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s does not exist.", name)),
+ diagnosticError != null ? diagnosticError : "Tool does not exist.",
+ success,
+ error,
+ responses);
+ continue;
}
+
+ try {
+ // Framework-owned injected args must win over model-provided values so hidden
+ // context such as tenant ids cannot be spoofed by a tool call payload.
+ mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
+ } catch (Exception e) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s execute failed.", name)),
+ e.getMessage(),
+ success,
+ error,
+ responses);
+ continue;
+ }
+
+ final Tool toolRef = tool;
+ final Map<String, Object> callArguments = mergedArguments;
+ DurableCallable<ToolResponse> callable =
+ new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "tool-call-" + id;
+ }
+
+ @Override
+ public Class<ToolResponse> getResultClass() {
+ return ToolResponse.class;
+ }
+
+ @Override
+ public ToolResponse call() throws Exception {
+ return toolRef.call(new ToolParameters(callArguments));
+ }
+ };
+ executions.add(new ToolCallExecution(id, name, callable));
+ }
+ return executions;
+ }
+
+ private static void executeParallel(
+ List<ToolCallExecution> executions,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses) {
+ List<DurableCallable<ToolResponse>> callables = new ArrayList<>(executions.size());
+ for (ToolCallExecution execution : executions) {
+ callables.add(execution.callable);
+ }
+ try {
+ List<Outcome<ToolResponse>> outcomes = ctx.durableExecuteAllAsync(callables);
+ for (int i = 0; i < outcomes.size(); i++) {
+ recordOutcome(executions.get(i), outcomes.get(i), success, error, responses);
+ }
+ } catch (Exception e) {
Review Comment:
Any throw out of `durableExecuteAllAsync` after partial progress marks every tool failed, including slots already durably finalized SUCCEEDED. The action is then marked completed, so those results never surface on a later run. Lost, not delayed.
The serial path just below degrades per tool; this one is all-or-nothing. Routes in besides the `Duration` conversion on `AgentExecutionOptions.java:82`: a `JsonProcessingException` out of `finalizeExecutedOutcomes` (`JavaRunnerContextImpl.java:211`), or a deserialization failure in `readTerminalOutcomeAt`.
Would recording the batch-level exception only against slots with no outcome yet work here, or is failing the whole set the intent?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -636,6 +721,110 @@ def wrapped_func(*a: Any, **kw: Any) -> Any:
return wrapped_func
+ def _call_matches(
+ self, current: _PersistedCallResult, call: DurableCall, args_digest: str
+ ) -> bool:
+ return current.function_id == call.id and current.args_digest == args_digest
+
+ def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
+ if current.exception_payload is not None:
+ return Outcome.failure(cloudpickle.loads(current.exception_payload))
+ if current.result_payload is None:
+ return Outcome.success(None)
+ return Outcome.success(cloudpickle.loads(current.result_payload))
+
+ def _callable_for_durable_call(self, call: DurableCall) -> Callable[[], Any]:
+ kwargs = call.kwargs or {}
+ return partial(call.func, *call.args, **kwargs)
+
+ def _prepare_batch_execution(self, calls: list[DurableCall]) -> _BatchExecutionPlan:
+ args_digest = ""
Review Comment:
The batch keys slots on `("tool-call-<id>", "")` (`tool_call_action.py:134`). The Python single-call path keys on `(_compute_function_id(func), args_digest)` (`:482-483`), and since the callable is `tool.call`, `_compute_function_id` returns the same string for every tool. Java has no such split: `RunnerContextImpl.java:604-605` uses `getId()` plus `""` for the single-call state machine too.
So a job that checkpoints mid-batch and restarts with `tool-call.parallel=false` replays cleanly on Java, while on Python `matchNextOrClearSubsequentCallResult` cannot match the persisted key, truncates the journal, and calls every completed tool again. Same flip that reaches the serial-path issue on `JavaRunnerContextImpl.java:193`.
The per-call `functionId` rename was headed for its own issue, so not asking for it here. What caught my eye is that the batch path adopted it while the single-call path did not, so the two disagree inside Python today. Is documenting the flip as not recoverable across a restart the right stopgap until that lands?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,6 +102,159 @@ private <T> T durableExecuteAsyncWithReconcile(
callable, reconcileCallable, () -> executeAsyncCallable(callable));
}
+ @Override
+ public <T> List<Outcome<T>> durableExecuteAllAsync(List<DurableCallable<T>> callables)
+ throws Exception {
+ if (callables.isEmpty()) {
+ return List.of();
+ }
+ if (durableExecutionContext == null) {
+ return executeAllWithoutDurableState(callables);
+ }
+
+ String argsDigest = "";
+ int base = durableExecutionContext.getCurrentCallIndex();
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base, argsDigest);
+
+ reservePendingBatchIfNeeded(callables, argsDigest, plan);
+
+ List<Outcome<T>> executed = executeOutcomeSuppliers(plan.suppliers);
+ finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+
+ advanceCallIndexBy(callables.size());
+ return plan.outcomes;
+ }
+
+ private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
+ List<DurableCallable<T>> callables, int base, String argsDigest) throws Exception {
+ BatchExecutionPlan<T> plan = new BatchExecutionPlan<>(callables.size());
+ for (int i = 0; i < callables.size(); i++) {
+ DurableCallable<T> callable = callables.get(i);
+ CallResult current = getCallResultAt(base + i);
+ if (current == null) {
+ markReservationStart(plan, i);
+ addExecutableCall(plan, i, callable::call);
+ continue;
+ }
+ if (!current.matches(callable.getId(), argsDigest)) {
+ clearCallResultsFromAndPersist(base + i);
+ plan.needsReservation = true;
+ plan.executionStart = i;
+ addExecutableCall(plan, i, callable::call);
+ appendRemainingExecutions(callables, plan, i + 1);
+ break;
+ }
+ if (current.isPending()) {
+ Callable<T> reconcileCallable = callable.reconciler();
+ Callable<T> executionCallable =
+ reconcileCallable != null ? reconcileCallable : callable::call;
+ addExecutableCall(plan, i, executionCallable);
+ } else {
+ plan.outcomes.add(
+ readTerminalOutcomeAt(
+ base + i, callable.getId(), argsDigest, callable.getResultClass()));
+ }
+ }
+ return plan;
+ }
+
+ private <T> void markReservationStart(BatchExecutionPlan<T> plan, int callIndex) {
+ plan.needsReservation = true;
+ if (plan.executionStart < 0) {
+ plan.executionStart = callIndex;
+ }
+ }
+
+ private <T> void addExecutableCall(
+ BatchExecutionPlan<T> plan, int callIndex, Callable<T> executionCallable) {
+ plan.outcomes.add(null);
+ plan.suppliers.add(executionCallable);
+ plan.executableCallIndexes.add(callIndex);
+ }
+
+ private <T> void appendRemainingExecutions(
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan, int startIndex) {
+ for (int i = startIndex; i < callables.size(); i++) {
+ DurableCallable<T> remaining = callables.get(i);
+ addExecutableCall(plan, i, remaining::call);
+ }
+ }
+
+ private <T> void reservePendingBatchIfNeeded(
+ List<DurableCallable<T>> callables, String argsDigest, BatchExecutionPlan<T> plan) {
+ if (!plan.needsReservation) {
+ return;
+ }
+ List<String> ids = new ArrayList<>();
+ for (DurableCallable<T> callable :
+ callables.subList(plan.executionStart, callables.size())) {
+ ids.add(callable.getId());
+ }
+ reservePendingBatch(ids, argsDigest);
Review Comment:
Reserving PENDING for every call in the batch, non-reconcilable ones included, is the intended shape, and it carries the guarantee that such a slot is re-executed on recovery. The batch path honours that at `:147-151`. The serial path can read those same slots and does not.
Before this PR the two never met. PENDING was written only by `durableExecuteWithReconcile` (`RunnerContextImpl.java:612`, `:618`) and the Python bridge's `_prepare_reconciler_execution`, both gated on a non-null `reconciler()`, and that method gates its own cache read on `!current.isPending()` at `:622`. The completion-only path never needed that gate, because non-reconcilable calls never wrote a PENDING slot. Now they do: `ToolCallAction.java:135` never overrides `reconciler()` and the interface default is `null`.
`CallResult.matches` compares only `functionId` and `argsDigest` (`CallResult.java:160-162`), so a reserved slot reads as a cache hit at `RunnerContextImpl.java:757`, both payloads are null, and `tryGetCachedResult` throws NPE on `return Optional.of(null)` at `:575`. `ToolCallAction.java:192` catches it, so the tool is reported failed rather than re-executed.
Two routes onto the serial path after a mid-batch crash, with the reserved slots still checkpointed. Setting `tool-call.parallel=false` and restarting is one. The other needs no config change: a trailing tool whose resource fails to resolve on the recovery run is dropped at `ToolCallAction.java:114`, `executions.size()` falls to 1, and the `> 1` guard at `:67` routes to serial.
Treating a PENDING slot as a miss in `matchNextOrClearSubsequentCallResult` / `tryGetCachedResult` would extend the same re-execute guarantee to that path. Would that be the right place for it, or is there a reason the serial read should keep seeing a hit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3864604546
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +293,158 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ parallelism = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_PARALLELISM)
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms > 0 else None
+ suppliers = [supplier for _, supplier in plan.suppliers]
+ batch_futures: list[Any | None] = [None] * len(suppliers)
+ started: list[bool] = [False] * len(suppliers)
+ try:
+ executed = yield from _execute_sliding_window_batch(
+ self._ctx.executor,
+ suppliers,
+ parallelism,
+ deadline,
+ timeout_ms,
+ batch_futures,
+ started,
+ )
+ except _BatchTimeoutError as exception:
+ executed = _collect_sliding_window_outcomes_on_timeout(
+ batch_futures, started, exception
+ )
+ return self._ctx._finalize_batch_execution(
+ self._calls, plan, started, executed
+ )
+
+
+class _BatchTimeoutError(TimeoutError):
+ """Raised when a durable batch exceeds its deadline."""
+
+
+def _mark_started_on_run(
+ supplier: Callable[[], Any], started: list[bool], index: int
+) -> Callable[[], Any]:
+ """Wrap a supplier so ``started[index]`` flips only when the worker truly runs.
+
+ A task queued in a saturated pool but cancelled before it executes keeps
+ ``started[index] == False``, so it is treated as never-run and stays
+ re-executable on recovery instead of being recorded as a timeout failure.
+ """
+
+ def _run() -> Any:
+ started[index] = True
+ return supplier()
+
+ return _run
+
+
+def _execute_sliding_window_batch(
+ executor: ThreadPoolExecutor,
+ suppliers: list[Any],
+ parallelism: int,
+ deadline: float | None,
+ timeout_ms: int,
+ futures: list[Any | None],
+ started: list[bool],
+) -> Any:
+ batch_size = len(suppliers)
+ if batch_size == 0:
+ return []
+
+ parallelism_limit = min(max(parallelism, 1), batch_size)
+ next_to_submit = 0
+ completed = 0
+ counted = [False] * batch_size
+
+ def in_flight() -> int:
+ return sum(
+ 1
+ for i in range(next_to_submit)
+ if futures[i] is not None and not futures[i].done()
+ )
+
+ while completed < batch_size:
+ if deadline is not None and time.monotonic() >= deadline:
+ timeout_message = (
+ f"Async durable batch execution timed out after {timeout_ms} ms"
+ )
+ raise _BatchTimeoutError(timeout_message)
+
+ while next_to_submit < batch_size and in_flight() < parallelism_limit:
+ index = next_to_submit
+ futures[index] = executor.submit(
+ _mark_started_on_run(suppliers[index], started, index)
+ )
+ next_to_submit += 1
+
+ for i in range(next_to_submit):
+ if not counted[i] and futures[i].done():
+ counted[i] = True
+ completed += 1
+
+ if completed < batch_size:
+ yield
+
+ return _collect_outcomes(futures)
+
+
+def _collect_sliding_window_outcomes_on_timeout(
+ futures: list[Any | None],
+ started: list[bool],
+ timeout_exception: BaseException,
+) -> list[Outcome]:
+ outcomes = []
+ for is_started, future in zip(started, futures, strict=True):
+ if not is_started or future is None:
+ outcomes.append(Outcome.failure(timeout_exception))
+ continue
+ if not future.done():
+ future.cancel()
+ if future.done() and not future.cancelled():
+ try:
+ outcomes.append(Outcome.success(future.result()))
+ except Exception as e:
+ outcomes.append(Outcome.failure(e))
+ else:
+ outcomes.append(Outcome.failure(timeout_exception))
+ return outcomes
+
+
+def _collect_outcomes(futures: list[Any]) -> list[Outcome]:
+ outcomes = []
+ for future in futures:
+ try:
+ outcomes.append(Outcome.success(future.result()))
+ except Exception as e: # noqa: PERF203
+ outcomes.append(Outcome.failure(e))
+ return outcomes
+
+
+def _collect_outcomes_on_timeout(
Review Comment:
removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5447994491
Apologies, I merged another PR that caused conflicts with the main branch again. Once @joeyutong and @weiqingy have reviewed and confirmed everything is correct, I will prioritize merging this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3675774615
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,6 +102,159 @@ private <T> T durableExecuteAsyncWithReconcile(
callable, reconcileCallable, () -> executeAsyncCallable(callable));
}
+ @Override
+ public <T> List<Outcome<T>> durableExecuteAllAsync(List<DurableCallable<T>> callables)
+ throws Exception {
+ if (callables.isEmpty()) {
+ return List.of();
+ }
+ if (durableExecutionContext == null) {
+ return executeAllWithoutDurableState(callables);
+ }
+
+ String argsDigest = "";
+ int base = durableExecutionContext.getCurrentCallIndex();
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base, argsDigest);
+
+ reservePendingBatchIfNeeded(callables, argsDigest, plan);
+
+ List<Outcome<T>> executed = executeOutcomeSuppliers(plan.suppliers);
+ finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+
+ advanceCallIndexBy(callables.size());
+ return plan.outcomes;
+ }
+
+ private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
+ List<DurableCallable<T>> callables, int base, String argsDigest) throws Exception {
+ BatchExecutionPlan<T> plan = new BatchExecutionPlan<>(callables.size());
+ for (int i = 0; i < callables.size(); i++) {
+ DurableCallable<T> callable = callables.get(i);
+ CallResult current = getCallResultAt(base + i);
+ if (current == null) {
+ markReservationStart(plan, i);
+ addExecutableCall(plan, i, callable::call);
+ continue;
+ }
+ if (!current.matches(callable.getId(), argsDigest)) {
+ clearCallResultsFromAndPersist(base + i);
+ plan.needsReservation = true;
+ plan.executionStart = i;
+ addExecutableCall(plan, i, callable::call);
+ appendRemainingExecutions(callables, plan, i + 1);
+ break;
+ }
+ if (current.isPending()) {
+ Callable<T> reconcileCallable = callable.reconciler();
+ Callable<T> executionCallable =
+ reconcileCallable != null ? reconcileCallable : callable::call;
+ addExecutableCall(plan, i, executionCallable);
+ } else {
+ plan.outcomes.add(
+ readTerminalOutcomeAt(
+ base + i, callable.getId(), argsDigest, callable.getResultClass()));
+ }
+ }
+ return plan;
+ }
+
+ private <T> void markReservationStart(BatchExecutionPlan<T> plan, int callIndex) {
+ plan.needsReservation = true;
+ if (plan.executionStart < 0) {
+ plan.executionStart = callIndex;
+ }
+ }
+
+ private <T> void addExecutableCall(
+ BatchExecutionPlan<T> plan, int callIndex, Callable<T> executionCallable) {
+ plan.outcomes.add(null);
+ plan.suppliers.add(executionCallable);
+ plan.executableCallIndexes.add(callIndex);
+ }
+
+ private <T> void appendRemainingExecutions(
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan, int startIndex) {
+ for (int i = startIndex; i < callables.size(); i++) {
+ DurableCallable<T> remaining = callables.get(i);
+ addExecutableCall(plan, i, remaining::call);
+ }
+ }
+
+ private <T> void reservePendingBatchIfNeeded(
+ List<DurableCallable<T>> callables, String argsDigest, BatchExecutionPlan<T> plan) {
+ if (!plan.needsReservation) {
+ return;
+ }
+ List<String> ids = new ArrayList<>();
+ for (DurableCallable<T> callable :
+ callables.subList(plan.executionStart, callables.size())) {
+ ids.add(callable.getId());
+ }
+ reservePendingBatch(ids, argsDigest);
Review Comment:
Yes, I'll mark it as 'no result read' during the read operation and include this in the next commit. I'm a bit tight on time at the moment, so I plan to push it over the weekend. We can revisit it then～
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3680736901
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -636,6 +721,110 @@ def wrapped_func(*a: Any, **kw: Any) -> Any:
return wrapped_func
+ def _call_matches(
+ self, current: _PersistedCallResult, call: DurableCall, args_digest: str
+ ) -> bool:
+ return current.function_id == call.id and current.args_digest == args_digest
+
+ def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
+ if current.exception_payload is not None:
+ return Outcome.failure(cloudpickle.loads(current.exception_payload))
+ if current.result_payload is None:
+ return Outcome.success(None)
+ return Outcome.success(cloudpickle.loads(current.result_payload))
+
+ def _callable_for_durable_call(self, call: DurableCall) -> Callable[[], Any]:
+ kwargs = call.kwargs or {}
+ return partial(call.func, *call.args, **kwargs)
+
+ def _prepare_batch_execution(self, calls: list[DurableCall]) -> _BatchExecutionPlan:
+ args_digest = ""
Review Comment:
Good catch — you're right that there’s still an upgrade cost here.
I agree we should split the broader semantics (“unique per-call id as recovery identity” / the same-args-different-result case) into a separate issue. For this PR I’ll revert the tool functionId back to the fixed "tool-call" on both Java and Python, so serial and batch stay consistent with the previous release and we don’t change recovery identity in this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3671213309
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +253,58 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ futures = [
+ self._ctx.tool_call_executor.submit(supplier) for _, supplier in plan.suppliers
+ ]
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms >= 0 else None
Review Comment:
`timeout_ms >= 0` makes a configured `0` mean "deadline is now" rather than "no timeout" — the first check in the loop fires and every batch fails with a `TimeoutError` before any tool can finish.
Java goes the other way, returning no deadline when `timeout.isZero() || timeout.isNegative()` (`ContinuationActionExecutor.java:255`), and the config table added in this PR documents it as "Non-positive disables it" (`configuration.md:137`). The `-1` default keeps this out of the way today, but `0` is the natural thing to write for "no timeout". Should this be `> 0`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3665375970
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,6 +102,159 @@ private <T> T durableExecuteAsyncWithReconcile(
callable, reconcileCallable, () -> executeAsyncCallable(callable));
}
+ @Override
+ public <T> List<Outcome<T>> durableExecuteAllAsync(List<DurableCallable<T>> callables)
+ throws Exception {
+ if (callables.isEmpty()) {
+ return List.of();
+ }
+ if (durableExecutionContext == null) {
+ return executeAllWithoutDurableState(callables);
+ }
+
+ String argsDigest = "";
+ int base = durableExecutionContext.getCurrentCallIndex();
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base, argsDigest);
+
+ reservePendingBatchIfNeeded(callables, argsDigest, plan);
+
+ List<Outcome<T>> executed = executeOutcomeSuppliers(plan.suppliers);
+ finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+
+ advanceCallIndexBy(callables.size());
+ return plan.outcomes;
+ }
+
+ private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
+ List<DurableCallable<T>> callables, int base, String argsDigest) throws Exception {
+ BatchExecutionPlan<T> plan = new BatchExecutionPlan<>(callables.size());
+ for (int i = 0; i < callables.size(); i++) {
+ DurableCallable<T> callable = callables.get(i);
+ CallResult current = getCallResultAt(base + i);
+ if (current == null) {
+ markReservationStart(plan, i);
+ addExecutableCall(plan, i, callable::call);
+ continue;
+ }
+ if (!current.matches(callable.getId(), argsDigest)) {
+ clearCallResultsFromAndPersist(base + i);
+ plan.needsReservation = true;
+ plan.executionStart = i;
+ addExecutableCall(plan, i, callable::call);
+ appendRemainingExecutions(callables, plan, i + 1);
+ break;
+ }
+ if (current.isPending()) {
+ Callable<T> reconcileCallable = callable.reconciler();
+ Callable<T> executionCallable =
+ reconcileCallable != null ? reconcileCallable : callable::call;
+ addExecutableCall(plan, i, executionCallable);
+ } else {
+ plan.outcomes.add(
+ readTerminalOutcomeAt(
+ base + i, callable.getId(), argsDigest, callable.getResultClass()));
+ }
+ }
+ return plan;
+ }
+
+ private <T> void markReservationStart(BatchExecutionPlan<T> plan, int callIndex) {
+ plan.needsReservation = true;
+ if (plan.executionStart < 0) {
+ plan.executionStart = callIndex;
+ }
+ }
+
+ private <T> void addExecutableCall(
+ BatchExecutionPlan<T> plan, int callIndex, Callable<T> executionCallable) {
+ plan.outcomes.add(null);
+ plan.suppliers.add(executionCallable);
+ plan.executableCallIndexes.add(callIndex);
+ }
+
+ private <T> void appendRemainingExecutions(
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan, int startIndex) {
+ for (int i = startIndex; i < callables.size(); i++) {
+ DurableCallable<T> remaining = callables.get(i);
+ addExecutableCall(plan, i, remaining::call);
+ }
+ }
+
+ private <T> void reservePendingBatchIfNeeded(
+ List<DurableCallable<T>> callables, String argsDigest, BatchExecutionPlan<T> plan) {
+ if (!plan.needsReservation) {
+ return;
+ }
+ List<String> ids = new ArrayList<>();
+ for (DurableCallable<T> callable :
+ callables.subList(plan.executionStart, callables.size())) {
+ ids.add(callable.getId());
+ }
+ reservePendingBatch(ids, argsDigest);
Review Comment:
Good catch! This is indeed a serious issue. I'll fix the problem where non‑reconcilable ones under the serial path contain pending slots. For recovery reruns, I'll use `finalizeCallAt `to handle the existing pending slots instead of creating a new one, which would otherwise cause another error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5436646964
The Python UTs are failing in `test_async_workers_carry_descriptive_names`. This test comes from #1028, which added descriptive names for Python async executor threads.
It looks like the rebase restored the pre-#1028 implementation of `create_async_thread_pool()`, dropping `_ASYNC_POOL_ID` and `thread_name_prefix` while retaining the test from `main`. Could you restore the #1028 implementation here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3671218164
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java:
##########
@@ -79,55 +103,160 @@ public static void processToolRequest(Event event, RunnerContext ctx) {
diagnosticError = e.getMessage();
}
- if (tool != null) {
- try {
- // Framework-owned injected args must win over model-provided values so hidden
- // context such as tenant ids cannot be spoofed by a tool call payload.
- mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
- ToolResponse response;
- final Tool toolRef = tool;
- final Map<String, Object> callArguments = mergedArguments;
- DurableCallable<ToolResponse> callable =
- new DurableCallable<>() {
- @Override
- public String getId() {
- return "tool-call";
- }
-
- @Override
- public Class<ToolResponse> getResultClass() {
- return ToolResponse.class;
- }
-
- @Override
- public ToolResponse call() throws Exception {
- return toolRef.call(new ToolParameters(callArguments));
- }
- };
- response =
- toolCallAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
- success.put(id, response.isSuccess());
- responses.put(id, response);
- if (!response.isSuccess() && response.getError() != null) {
- error.put(id, response.getError());
- }
- } catch (Exception e) {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s execute failed.", name)));
- error.put(id, e.getMessage());
- }
- } else {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s does not exist.", name)));
- error.put(id, diagnosticError != null ? diagnosticError : "Tool does not exist.");
+ if (tool == null) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s does not exist.", name)),
+ diagnosticError != null ? diagnosticError : "Tool does not exist.",
+ success,
+ error,
+ responses);
+ continue;
}
+
+ try {
+ // Framework-owned injected args must win over model-provided values so hidden
+ // context such as tenant ids cannot be spoofed by a tool call payload.
+ mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
+ } catch (Exception e) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s execute failed.", name)),
+ e.getMessage(),
+ success,
+ error,
+ responses);
+ continue;
+ }
+
+ final Tool toolRef = tool;
+ final Map<String, Object> callArguments = mergedArguments;
+ DurableCallable<ToolResponse> callable =
+ new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "tool-call-" + id;
+ }
+
+ @Override
+ public Class<ToolResponse> getResultClass() {
+ return ToolResponse.class;
+ }
+
+ @Override
+ public ToolResponse call() throws Exception {
+ return toolRef.call(new ToolParameters(callArguments));
+ }
+ };
+ executions.add(new ToolCallExecution(id, name, callable));
+ }
+ return executions;
+ }
+
+ private static void executeParallel(
+ List<ToolCallExecution> executions,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses) {
+ List<DurableCallable<ToolResponse>> callables = new ArrayList<>(executions.size());
+ for (ToolCallExecution execution : executions) {
+ callables.add(execution.callable);
+ }
+ try {
+ List<Outcome<ToolResponse>> outcomes = ctx.durableExecuteAllAsync(callables);
+ for (int i = 0; i < outcomes.size(); i++) {
+ recordOutcome(executions.get(i), outcomes.get(i), success, error, responses);
+ }
+ } catch (Exception e) {
Review Comment:
Handling it inside `durableExecuteAllAsync` so the caller only reads outcomes is a cleaner seam, agreed. Where would you draw the line for failures that aren't a tool's fault? A state-store write failure mapped into N per-tool outcomes would let the action complete normally, with the agent carrying on as though "all tools failed" were a real result. Would it be worth letting those keep propagating, and mapping only per-call execution failures into outcomes?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -636,6 +721,110 @@ def wrapped_func(*a: Any, **kw: Any) -> Any:
return wrapped_func
+ def _call_matches(
+ self, current: _PersistedCallResult, call: DurableCall, args_digest: str
+ ) -> bool:
+ return current.function_id == call.id and current.args_digest == args_digest
+
+ def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
+ if current.exception_payload is not None:
+ return Outcome.failure(cloudpickle.loads(current.exception_payload))
+ if current.result_payload is None:
+ return Outcome.success(None)
+ return Outcome.success(cloudpickle.loads(current.result_payload))
+
+ def _callable_for_durable_call(self, call: DurableCall) -> Callable[[], Any]:
+ kwargs = call.kwargs or {}
+ return partial(call.func, *call.args, **kwargs)
+
+ def _prepare_batch_execution(self, calls: list[DurableCall]) -> _BatchExecutionPlan:
+ args_digest = ""
Review Comment:
Happy either way, but I think two separable things are bundled under `functionId`, and only one of them is the piece we agreed to defer.
The narrow one is the divergence this PR introduces, which feels in scope here. `_build_executions` already mints `DurableCall(id=f"tool-call-{call_id}")` (`tool_call_action.py:134`), but `_execute_sequentially` passes only `call.func` (`:174`, `:180`), so the id is dropped and the runtime falls back to `_compute_function_id` → `module.qualname`, the same string for every tool. Java has no such split — `buildCallables` mints one callable and both paths use it. Threading the id that's already built through the serial path looks contained.
The broader one is using a unique per-call id as recovery identity — the "same args, different result" case we split out to its own issue. Worth flagging that this PR already crosses that line on Java: `getId()` returns the constant `"tool-call"` on `main`, and `"tool-call-" + id` here, for the serial path as well as the batch.
That carries an upgrade cost either way. An action checkpointed mid-tool-loop on the current release and restored on this build recorded `functionId="tool-call"`, so `matchNextOrClearSubsequentCallResult` (`RunnerContextImpl.java:654`) won't match, clears from the current index, and the already-completed tool calls run again — surfaced in the log as "Non-deterministic call detected", which would point an operator at the wrong cause.
So my leaning would be to keep the serial-path id threading here as a parity fix and let the separate issue carry the semantics change. If you'd rather land both together, that works too — the piece worth writing down either way is the cross-version recovery behaviour. What's your read?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java:
##########
@@ -240,12 +397,61 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) {
return supplier.get();
}
+ @Override
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ java.time.Duration timeout) {
+ executeAllAsyncCallCount++;
+ executeAllAsyncBatchSizes.add(suppliers.size());
+ if (useTimeoutCollection) {
+ return collectTimedOutOutcomes(suppliers);
+ }
+ List<Outcome<T>> outcomes = new java.util.ArrayList<>(suppliers.size());
+ for (Callable<T> supplier : suppliers) {
+ try {
+ outcomes.add(Outcome.success(supplier.call()));
+ } catch (Exception e) {
+ outcomes.add(Outcome.failure(e));
+ }
+ }
+ return outcomes;
+ }
+
+ private <T> List<Outcome<T>> collectTimedOutOutcomes(List<Callable<T>> suppliers) {
Review Comment:
Sounds good, and sequencing it after the config fix makes sense. One gap the e2e won't close though: the stub at `:404` ignores the `timeout` argument it's handed, so that assertion passes regardless of the deadline. Worth tightening at the same time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "pltbkd (via GitHub)" <gi...@apache.org>.
pltbkd commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5166600871
Thanks @da-daken for proposing this. The thorough code review from @weiqingy already covers a lot — thanks for that. On top of it, I'd like to raise a few concerns of my own.
1. DurableCall vs Java's DurableCallable — not aligned
The new Python DurableCall and Java's DurableCallable are not aligned: Java uses the caller-declared getId() for recovery matching, while Python never consumes any caller-supplied identity and derives it from func + args instead. This seems to conflict with the requirement that new APIs stay semantically aligned across languages (and the class names differ too).
That said, this is essentially a legacy gap, and this PR actually partly closes it. From a semantic-alignment standpoint it's fine to keep the current state (including the removal of the id field) for v1; I'd just suggest we track it as a tech-debt follow-up to bring Python's durable identity back in line with Java's getId() contract — including aligning the class names.
2. Timeout — confirm reconcile impact before merge; and subagents will need this too
The timeout here is a mechanism on durable_execute, but the framework has no way to actually cancel a running callable. If the callable has side effects, then after a timeout the framework records the state as failed while the side effects may still happen — leaving flink-agents state inconsistent with the actual side effects. The exception path is also long, and the exception type is lost on recovery replay (TimeoutException becomes a bare RuntimeException).
I'm not asking to remove the timeout here. But before merging, I'd like to confirm the interaction with reconciler: reconciler only fires on PENDING slots, and timeout finalizes the slot to FAILED — so after a timeout, the reconcile channel looks closed. Is that the intended behavior?
Also worth noting: subagents are expected to use the durableExecuteAllAsync API too. If the mechanism can't properly support side-effecting callables (cancel + post-timeout state semantics), that gap will need to be filled soon — not just for tool calls.
3. Do we really need two thread pools?
The motivation in the discussion for a dedicated pool was to avoid one ToolRequestEvent exhausting the global pool and affecting other keys. But if every record carries multiple parallel tool calls (which is the typical case), the tool pool gets saturated just the same — the problem simply moves to the other pool. Meanwhile, the default thread count is based on CPU cores, a sizing heuristic meant for compute-bound work; tool calls are I/O bound, so adding another pool of the same size only increases CPU and memory pressure.
I think max-parallelism (a single shared pool + a cap on how many threads one record/batch can occupy) may be more suitable. It can be a follow-up optimization. But if that's the direction, I'd suggest avoiding introducing a separate tool-call.num-async-threads config now, since it would likely need to be removed or reworked when we move to that model.
4. Config surface — potential redundancy given the above
Following from point 3: if max-parallelism turns out to be the more suitable direction, tool-call.num-async-threads looks redundant — the single pool is already sized by num-async-threads. At the same time, the new tool-call.parallel boolean switch could simply be folded into parallelism = 1 (serial behavior). So the current three-option surface looks like it carries config we'd likely revisit or collapse shortly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3793394265
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +294,136 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ parallelism = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_PARALLELISM)
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms > 0 else None
+ suppliers = [supplier for _, supplier in plan.suppliers]
+ batch_futures: list[Any | None] = [None] * len(suppliers)
+ try:
+ executed = yield from _execute_sliding_window_batch(
+ self._ctx.executor,
+ suppliers,
+ parallelism,
+ deadline,
+ timeout_ms,
+ batch_futures,
+ plan.submitted,
+ )
+ except _BatchTimeoutError as exception:
+ executed = _collect_sliding_window_outcomes_on_timeout(
+ batch_futures, plan.submitted, exception
+ )
+ return self._ctx._finalize_batch_execution(self._calls, plan, executed)
+
+
+class _BatchTimeoutError(TimeoutError):
+ """Raised when a durable batch exceeds its deadline."""
+
+
+def _execute_sliding_window_batch(
+ executor: ThreadPoolExecutor,
+ suppliers: list[Any],
+ parallelism: int,
+ deadline: float | None,
+ timeout_ms: int,
+ futures: list[Any | None],
+ submitted: list[bool],
+) -> Any:
+ batch_size = len(suppliers)
+ if batch_size == 0:
+ return []
+
+ parallelism_limit = min(max(parallelism, 1), batch_size)
+ next_to_submit = 0
+ completed = 0
+ counted = [False] * batch_size
+
+ def in_flight() -> int:
+ return sum(
+ 1
+ for i in range(next_to_submit)
+ if futures[i] is not None and not futures[i].done()
+ )
+
+ while completed < batch_size:
+ if deadline is not None and time.monotonic() >= deadline:
+ timeout_message = (
+ f"Async durable batch execution timed out after {timeout_ms} ms"
+ )
+ raise _BatchTimeoutError(timeout_message)
+
+ while next_to_submit < batch_size and in_flight() < parallelism_limit:
+ futures[next_to_submit] = executor.submit(suppliers[next_to_submit])
+ submitted[next_to_submit] = True
+ next_to_submit += 1
+
+ for i in range(next_to_submit):
+ if not counted[i] and futures[i].done():
+ counted[i] = True
+ completed += 1
+
+ if completed < batch_size:
+ yield
+
+ return _collect_outcomes(futures)
+
+
+def _collect_sliding_window_outcomes_on_timeout(
+ futures: list[Any | None],
+ submitted: list[bool],
+ timeout_exception: BaseException,
+) -> list[Outcome]:
+ outcomes = []
+ for is_submitted, future in zip(submitted, futures, strict=True):
+ if not is_submitted or future is None:
+ outcomes.append(Outcome.failure(timeout_exception))
+ continue
+ if not future.done():
+ future.cancel()
+ if future.done() and not future.cancelled():
Review Comment:
Thanks, this closes the never-submitted case. I think there's one more case in the same timeout path though, and I'm curious how you'd want to handle it.
`submitted[i]` is set at `:364` the moment `executor.submit(...)` returns at `:363`, so it really means "handed to the pool" rather than "actually ran". At the deadline, `future.cancel()` at `:389` succeeds for anything still sitting in the queue, which stops it from ever running. Line `:390` even spots this (`future.done() and not future.cancelled()`), but the slot still falls through to `Outcome.failure(timeout_exception)` at `:396`. Since `submitted[i]` is `True`, `_finalize_batch_execution:1017` hands it to `finalizeCallAt` at `:1025` and it gets recorded as FAILED. After a restart, `_prepare_batch_execution:984-985` replays that failure rather than running the tool.
This only shows up when `tool-call.batch.timeout.ms` is set, since `:306` only builds a deadline when it's positive. With it on, you'd get there through the busy-pool case you describe at `core_options.py:257-264`: the pool is `num-async-threads` (default `2 x cores`) shared per subtask, `tool-call.parallelism` defaults to `cores`, so once the pool fills up `submit` starts queueing, and any queued slot at the deadline lands here.
One option would be to set the flag as the first line inside the supplier wrapper instead (`:363-364` here, `java21/ContinuationActionExecutor.java:210-219` on the Java side). That would keep both languages saying the same thing, since cancelling a `CompletableFuture` doesn't stop an already-queued supplier from running (`java21/ContinuationActionExecutor.java:301`). What do you think the flag should mean, handed to the pool or actually started?
##########
python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py:
##########
@@ -415,3 +594,458 @@ def collect_kwargs(**kwargs: Any) -> dict[str, Any]:
_close_runner_context(ctx)
assert result == {}
+
+
+def test_flink_runner_context_durable_execute_all_async_runs_calls_in_parallel() -> None:
+ j_runner_context = _FakeJavaRunnerContext()
+ config = AgentConfiguration(
+ {"tool-call.batch.timeout.ms": -1, "tool-call.parallelism": 3}
+ )
+ ctx = _create_runner_context(j_runner_context, config=config, executor_workers=3)
+ sleep_seconds = 0.2
+
+ def slow_call(value: str) -> str:
+ time.sleep(sleep_seconds)
+ return value
+
+ try:
+ start = time.perf_counter()
+ outcomes = _run_async(
+ ctx.durable_execute_all_async(
+ [
+ _durable_call(slow_call, "one"),
+ _durable_call(slow_call, "two"),
+ _durable_call(slow_call, "three"),
+ ]
+ )
+ )
+ elapsed = time.perf_counter() - start
+ finally:
+ _close_runner_context(ctx)
+
+ assert [outcome.value for outcome in outcomes] == ["one", "two", "three"]
+ assert elapsed < sleep_seconds * 2.5
Review Comment:
nit: with `sleep_seconds = 0.2` (`:605`), 3 calls and parallelism 3, moving to `2.5` shifts the threshold from 400 ms to 500 ms. That gives 100 ms more room for jitter, but it also halves the gap below the fully-serial 600 ms, from 200 ms down to 100 ms. The check also only ever caught fully serial execution, since 2 parallel plus 1 serial comes in around 400 ms and passes either way.
A `threading.Barrier(3)` awaited inside `slow_call` would make it deterministic instead. If execution went serial, the first call would block, the barrier would break, and the existing value check at `:626` would fail on its own, so the timing assert could go away. This one did go red on `ut-python [macos]` at the previous head `204eb7a3`. Would a barrier be worth it here?
##########
docs/content/docs/operations/configuration.md:
##########
@@ -131,9 +131,11 @@ Here is the list of all built-in core configuration options.
| `max-retries` | 3 | int | Number of retries when using `ErrorHandlingStrategy.RETRY`. |
| `retry-wait-interval` | 1 | int | Base wait interval in seconds between retries when using `ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the connection name. |
| `chat.async` | true | boolean | Whether chat asynchronously for built-in chat action. |
-| `tool-call.async` | true | boolean | Whether process tool call for built-in tool call action. |
+| `tool-call.async` | true | boolean | Whether the built-in tool-call action runs each tool via durable async execution. |
+| `tool-call.parallelism` | os cpu count | int | In-flight concurrency for tool calls from one `ToolRequestEvent` batch when `tool-call.async` is enabled. `1` runs tools serially; values greater than `1` run a parallel durable batch with a sliding window of at most that many concurrent tool calls. On **Java**, concurrent in-batch execution requires **JDK 21+** (Continuation API); below JDK 21 the batch still runs but tool calls execute serially. **Python** uses the shared async `ThreadPoolExecutor` and runs batches concurrently regardless of JDK version. Increases in-flight external calls; after failover, unfinished tools may be submitted again — side-effecting tools should be idempotent or provide a reconciler. {{< hint warning >}}**Default is parallel** (`os cpu count`). Chat, RAG, and tool batches share one `num-async-threads` pool **per operator subtask** (all keys on that subtask). Built-in actions for a single key run one at a time, so chat
and a tool batch on the **same key** do not overlap in the usual chat → tool path; delay shows up mainly **across keys** on the same subtask. With defaults (`num-async-threads = 2× cores`, `tool-call.parallelism = cores`), one batch can use up to half the pool; several busy keys can still saturate it. Lower this value or increase `num-async-threads` on hot subtasks. {{< /hint >}} |
+| `tool-call.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall timeout for one parallel tool-call batch. Non-positive disables it. On timeout, completed slots keep their outcome and unfinished slots fail. Timeout cancellation is best-effort; external side effects from unfinished tool calls may still complete, so side-effecting tools should be idempotent or provide a reconciler. On **Java**, only enforced on **JDK 21+**; on JDK 11 the batch fallback ignores this setting and runs to completion serially. **Python** enforces the deadline in the batch await loop. |
Review Comment:
Now that timed-out slots that never got submitted stay PENDING instead of failing, a few places still describe the old behaviour:
- this row: "On timeout, completed slots keep their outcome and unfinished slots fail."
- `docs/content/docs/development/tool_use.md:317-318`, the same sentence.
- `AgentExecutionOptions.java:78-79`, "When the deadline elapses, unfinished slots are failed", which sits right above the new thread-reclamation text at `:81-87`.
PENDING versus FAILED is what decides whether a timed-out tool runs again after a restart, so it seems worth keeping these in step with the code. Could the three be updated together?
nit: the thread-reclamation note you added went into `AgentExecutionOptions.java:81-87` and `core_options.py:272-280`, but not into either doc page. The `tool-call.parallelism` row at `:135` keeps its capacity warning in a `{{< hint warning >}}` block, if you wanted somewhere to put it.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -477,7 +599,7 @@ protected <T> Optional<T> tryGetCachedResult(
} else if (resultPayload != null) {
return Optional.of(OBJECT_MAPPER.readValue(resultPayload, resultClass));
} else {
- return Optional.of(null);
+ return Optional.empty();
Review Comment:
Happy to leave this out of the PR. One thing on the reasoning though, in case it matters later.
The PENDING check is real (`:789-797`), but it's the only check. `matchNextOrClearSubsequentCallResult` returns a hit at `:804-806` once `matches(functionId, argsDigest)` passes and the slot isn't pending, and nothing looks at the payloads. So a SUCCEEDED slot with a null result payload still gets through, and reaches `Optional.of(null)` at `:606`.
That's also how a `null` return gets stored. `serializeDurableResult(null)` returns `null` rather than a JSON `null` (`:691-693`), so a durable call that legitimately returns `null` is saved as SUCCEEDED with no payload, and NPEs on replay. `ToolCallAction` is fine, since the per-execution `catch` at `ToolCallAction.java:187-197` turns it into a tool failure. The case that's actually exposed is a user action calling `durableExecute` directly.
`main` carries the identical line (`RunnerContextImpl.java:618` there), so this PR isn't making anything worse, which is why I think deferring is the right call.
Worth an issue though, so it doesn't get lost. The same class already handles this case elsewhere: `readTerminalOutcomeAt` at `:566-568` does `if (callResult.getResultPayload() == null) return Outcome.success(null);`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5440976765
Hi @weiqingy, the latest three issues have been committed and I've rebased to resolve all conflicts.
@joeyutong, I handled many conflicts in ToolCallAction – would you mind reviewing to check if there are any merge errors?
@wenjin272 I've also added the missing PR: [#1028](https://github.com/apache/flink-agents/pull/1028%5Breference:1%5D)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3792073456
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -332,6 +352,23 @@ public static DurableExecutionException fromException(Exception e) {
}
public Exception toException() {
+ if (exceptionClass == null) {
+ return new RuntimeException(message);
+ }
+ try {
+ Class<?> clazz = Class.forName(exceptionClass);
+ if (Exception.class.isAssignableFrom(clazz)) {
+ @SuppressWarnings("unchecked")
+ Class<? extends Exception> exceptionClazz = (Class<? extends Exception>) clazz;
+ try {
+ return exceptionClazz.getConstructor(String.class).newInstance(message);
+ } catch (NoSuchMethodException ignored) {
+ return exceptionClazz.getConstructor().newInstance();
Review Comment:
Good catch. I verified this issue in `testDurableExecutionExceptionUsesContextClassLoader` and confirmed that it is indeed problematic. I have added a fix in the new commit to properly load using the user thread's classLoader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3864600818
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +293,158 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ parallelism = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_PARALLELISM)
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms > 0 else None
+ suppliers = [supplier for _, supplier in plan.suppliers]
+ batch_futures: list[Any | None] = [None] * len(suppliers)
+ started: list[bool] = [False] * len(suppliers)
+ try:
+ executed = yield from _execute_sliding_window_batch(
+ self._ctx.executor,
+ suppliers,
+ parallelism,
+ deadline,
+ timeout_ms,
+ batch_futures,
+ started,
+ )
+ except _BatchTimeoutError as exception:
+ executed = _collect_sliding_window_outcomes_on_timeout(
+ batch_futures, started, exception
+ )
+ return self._ctx._finalize_batch_execution(
+ self._calls, plan, started, executed
+ )
+
+
+class _BatchTimeoutError(TimeoutError):
+ """Raised when a durable batch exceeds its deadline."""
+
+
+def _mark_started_on_run(
+ supplier: Callable[[], Any], started: list[bool], index: int
+) -> Callable[[], Any]:
+ """Wrap a supplier so ``started[index]`` flips only when the worker truly runs.
+
+ A task queued in a saturated pool but cancelled before it executes keeps
+ ``started[index] == False``, so it is treated as never-run and stays
+ re-executable on recovery instead of being recorded as a timeout failure.
+ """
+
+ def _run() -> Any:
+ started[index] = True
+ return supplier()
+
+ return _run
+
+
+def _execute_sliding_window_batch(
+ executor: ThreadPoolExecutor,
+ suppliers: list[Any],
+ parallelism: int,
+ deadline: float | None,
+ timeout_ms: int,
+ futures: list[Any | None],
+ started: list[bool],
+) -> Any:
+ batch_size = len(suppliers)
+ if batch_size == 0:
+ return []
+
+ parallelism_limit = min(max(parallelism, 1), batch_size)
+ next_to_submit = 0
+ completed = 0
+ counted = [False] * batch_size
+
+ def in_flight() -> int:
+ return sum(
+ 1
+ for i in range(next_to_submit)
+ if futures[i] is not None and not futures[i].done()
+ )
+
+ while completed < batch_size:
+ if deadline is not None and time.monotonic() >= deadline:
+ timeout_message = (
+ f"Async durable batch execution timed out after {timeout_ms} ms"
+ )
+ raise _BatchTimeoutError(timeout_message)
+
+ while next_to_submit < batch_size and in_flight() < parallelism_limit:
+ index = next_to_submit
+ futures[index] = executor.submit(
+ _mark_started_on_run(suppliers[index], started, index)
+ )
+ next_to_submit += 1
+
+ for i in range(next_to_submit):
+ if not counted[i] and futures[i].done():
+ counted[i] = True
+ completed += 1
+
+ if completed < batch_size:
+ yield
+
+ return _collect_outcomes(futures)
+
+
+def _collect_sliding_window_outcomes_on_timeout(
+ futures: list[Any | None],
+ started: list[bool],
+ timeout_exception: BaseException,
+) -> list[Outcome]:
+ outcomes = []
+ for is_started, future in zip(started, futures, strict=True):
+ if not is_started or future is None:
Review Comment:
Good catch! Similarly, I observed that under OpenJDK, when a worker dequeues a task, AsyncSupply.run() checks whether d.result == null. After cancel is invoked, tasks already queued will not be executed. This behavior does not affect other JDK versions either, so I also removed that check on the Java side, allowing execution to fall through to the cancel method instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3676216535
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +253,58 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ futures = [
+ self._ctx.tool_call_executor.submit(supplier) for _, supplier in plan.suppliers
+ ]
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms >= 0 else None
Review Comment:
Thank you for the careful review～ I'll fix it in the next commit, changing it to `> 0`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3837386976
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +293,158 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ parallelism = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_PARALLELISM)
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms > 0 else None
+ suppliers = [supplier for _, supplier in plan.suppliers]
+ batch_futures: list[Any | None] = [None] * len(suppliers)
+ started: list[bool] = [False] * len(suppliers)
+ try:
+ executed = yield from _execute_sliding_window_batch(
+ self._ctx.executor,
+ suppliers,
+ parallelism,
+ deadline,
+ timeout_ms,
+ batch_futures,
+ started,
+ )
+ except _BatchTimeoutError as exception:
+ executed = _collect_sliding_window_outcomes_on_timeout(
+ batch_futures, started, exception
+ )
+ return self._ctx._finalize_batch_execution(
+ self._calls, plan, started, executed
+ )
+
+
+class _BatchTimeoutError(TimeoutError):
+ """Raised when a durable batch exceeds its deadline."""
+
+
+def _mark_started_on_run(
+ supplier: Callable[[], Any], started: list[bool], index: int
+) -> Callable[[], Any]:
+ """Wrap a supplier so ``started[index]`` flips only when the worker truly runs.
+
+ A task queued in a saturated pool but cancelled before it executes keeps
+ ``started[index] == False``, so it is treated as never-run and stays
+ re-executable on recovery instead of being recorded as a timeout failure.
+ """
+
+ def _run() -> Any:
+ started[index] = True
+ return supplier()
+
+ return _run
+
+
+def _execute_sliding_window_batch(
+ executor: ThreadPoolExecutor,
+ suppliers: list[Any],
+ parallelism: int,
+ deadline: float | None,
+ timeout_ms: int,
+ futures: list[Any | None],
+ started: list[bool],
+) -> Any:
+ batch_size = len(suppliers)
+ if batch_size == 0:
+ return []
+
+ parallelism_limit = min(max(parallelism, 1), batch_size)
+ next_to_submit = 0
+ completed = 0
+ counted = [False] * batch_size
+
+ def in_flight() -> int:
+ return sum(
+ 1
+ for i in range(next_to_submit)
+ if futures[i] is not None and not futures[i].done()
+ )
+
+ while completed < batch_size:
+ if deadline is not None and time.monotonic() >= deadline:
+ timeout_message = (
+ f"Async durable batch execution timed out after {timeout_ms} ms"
+ )
+ raise _BatchTimeoutError(timeout_message)
+
+ while next_to_submit < batch_size and in_flight() < parallelism_limit:
+ index = next_to_submit
+ futures[index] = executor.submit(
+ _mark_started_on_run(suppliers[index], started, index)
+ )
+ next_to_submit += 1
+
+ for i in range(next_to_submit):
+ if not counted[i] and futures[i].done():
+ counted[i] = True
+ completed += 1
+
+ if completed < batch_size:
+ yield
+
+ return _collect_outcomes(futures)
+
+
+def _collect_sliding_window_outcomes_on_timeout(
+ futures: list[Any | None],
+ started: list[bool],
+ timeout_exception: BaseException,
+) -> list[Outcome]:
+ outcomes = []
+ for is_started, future in zip(started, futures, strict=True):
+ if not is_started or future is None:
+ outcomes.append(Outcome.failure(timeout_exception))
+ continue
+ if not future.done():
+ future.cancel()
+ if future.done() and not future.cancelled():
+ try:
+ outcomes.append(Outcome.success(future.result()))
+ except Exception as e:
+ outcomes.append(Outcome.failure(e))
+ else:
+ outcomes.append(Outcome.failure(timeout_exception))
+ return outcomes
+
+
+def _collect_outcomes(futures: list[Any]) -> list[Outcome]:
+ outcomes = []
+ for future in futures:
+ try:
+ outcomes.append(Outcome.success(future.result()))
+ except Exception as e: # noqa: PERF203
+ outcomes.append(Outcome.failure(e))
+ return outcomes
+
+
+def _collect_outcomes_on_timeout(
Review Comment:
nit: `_collect_outcomes_on_timeout` is added by this PR and has no callers anywhere. `_collect_sliding_window_outcomes_on_timeout` at `:399` superseded it. Worth deleting before merge.
##########
runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java:
##########
@@ -195,6 +208,357 @@ void testDurableExecuteAsyncReconcilableReconcileExceptionPersistsFailure() thro
assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
}
+ @Test
+ void testDurableExecuteAsyncCompletionOnlyReExecutesPendingSlot() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(CallResult.pending("tool-call", ""));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>("tool-call", String.class, () -> "recovered");
+
+ String result = context.durableExecuteAsync(callable);
+
+ assertEquals("recovered", result);
+ assertEquals(1, callable.getCallCount());
+ assertEquals(1, executor.getExecuteAsyncCallCount());
+ assertEquals(1, persistCallCount.get());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ CallResult persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults().get(0);
+ assertTrue(persisted.isSuccess());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncInitialBatchPersistsOutcomes() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>("batch-1", String.class, () -> "one");
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>("batch-2", String.class, () -> "two");
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(first, second));
+
+ assertEquals("one", outcomes.get(0).getValue());
+ assertEquals("two", outcomes.get(1).getValue());
+ assertEquals(1, executor.getExecuteAllAsyncCallCount());
+ assertEquals(List.of(2), executor.getExecuteAllAsyncBatchSizes());
+ assertEquals(1, first.getCallCount());
+ assertEquals(1, second.getCallCount());
+ assertEquals(3, persistCallCount.get());
+ assertEquals(2, context.getDurableExecutionContext().getCurrentCallIndex());
+ List<CallResult> persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults();
+ assertEquals(2, persisted.size());
+ assertEquals("batch-1", persisted.get(0).getFunctionId());
+ assertTrue(persisted.get(0).isSuccess());
+ assertEquals("batch-2", persisted.get(1).getFunctionId());
+ assertTrue(persisted.get(1).isSuccess());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncReconcilesPendingSlot() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(CallResult.pending("batch-1", ""));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestReconcilableCallable<String> callable =
+ new TestReconcilableCallable<>(
+ "batch-1",
+ String.class,
+ () -> fail("call should not be executed"),
+ () -> "recovered");
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertEquals("recovered", outcomes.get(0).getValue());
+ assertEquals(0, callable.getCallCount());
+ assertEquals(1, callable.getReconcileCount());
+ assertEquals(1, executor.getExecuteAllAsyncCallCount());
+ assertEquals(1, persistCallCount.get());
+ assertTrue(actionState.getCallResults().get(0).isSuccess());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncRecoversPartialFinalizedBatch() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(
+ 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", ""));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>(
+ "batch-1", String.class, () -> fail("cached slot should not execute"));
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2", String.class, () -> fail("cached slot should not execute"));
+ TestDurableCallable<String> third =
+ new TestDurableCallable<>("batch-3", String.class, () -> "fresh-three");
+
+ List<Outcome<String>> outcomes =
+ context.durableExecuteAllAsync(List.of(first, second, third));
+
+ assertEquals("cached-one", outcomes.get(0).getValue());
+ assertEquals("cached-two", outcomes.get(1).getValue());
+ assertEquals("fresh-three", outcomes.get(2).getValue());
+ assertEquals(0, first.getCallCount());
+ assertEquals(0, second.getCallCount());
+ assertEquals(1, third.getCallCount());
+ assertEquals(1, executor.getExecuteAllAsyncCallCount());
+ assertEquals(List.of(1), executor.getExecuteAllAsyncBatchSizes());
+ assertEquals("batch-3", actionState.getCallResults().get(2).getFunctionId());
+ assertTrue(actionState.getCallResults().get(2).isSuccess());
+ assertEquals(1, persistCallCount.get());
+ assertEquals(3, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncReturnsCachedFailureOutcome() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(
+ new CallResult(
+ "batch-1",
+ "",
+ null,
+ OBJECT_MAPPER.writeValueAsBytes(
+ RunnerContextImpl.DurableExecutionException.fromException(
+ new IllegalStateException("cached failure")))));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>(
+ "batch-1", String.class, () -> fail("cached slot should not execute"));
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertTrue(outcomes.get(0).isFailure());
+ assertInstanceOf(IllegalStateException.class, outcomes.get(0).getError());
+ assertTrue(outcomes.get(0).getError().getMessage().contains("cached failure"));
+ assertEquals(0, callable.getCallCount());
+ assertEquals(0, executor.getExecuteAllAsyncCallCount());
+ assertEquals(0, persistCallCount.get());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncReturnsDeserializeFailureAsOutcome() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(
+ new CallResult(
+ "batch-1",
+ "",
+ "not-valid-json".getBytes(java.nio.charset.StandardCharsets.UTF_8),
+ null));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>(
+ "batch-1", String.class, () -> fail("cached slot should not execute"));
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertTrue(outcomes.get(0).isFailure());
+ assertInstanceOf(JsonProcessingException.class, outcomes.get(0).getError());
+ assertEquals(0, callable.getCallCount());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncPassesParallelismFromConfig() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ ((Configuration) context.getConfig()).set(AgentExecutionOptions.TOOL_CALL_PARALLELISM, 4);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>("batch-1", String.class, () -> "ok");
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertEquals("ok", outcomes.get(0).getValue());
+ assertEquals(4, executor.getLastExecuteAllAsyncMaxParallelism());
+ executor.close();
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncTimeoutKeepsCompletedOutcomes() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ executor.setUseTimeoutCollection(true);
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ ((Configuration) context.getConfig())
+ .set(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS, 100L);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>("batch-1", String.class, () -> "fast");
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2",
+ String.class,
+ () -> {
+ Thread.sleep(200);
+ return "slow";
+ });
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(first, second));
+
+ assertEquals("fast", outcomes.get(0).getValue());
+ assertTrue(outcomes.get(1).isFailure());
+ assertInstanceOf(TimeoutException.class, outcomes.get(1).getError());
+ assertEquals(Duration.ofMillis(100), executor.getLastExecuteAllAsyncTimeout());
+ assertEquals(1, first.getCallCount());
+ assertEquals(1, second.getCallCount());
+ List<CallResult> persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults();
+ assertTrue(persisted.get(0).isSuccess());
+ assertTrue(persisted.get(1).isFailure());
+ assertEquals(2, context.getDurableExecutionContext().getCurrentCallIndex());
+ executor.close();
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncTimeoutLeavesUnsubmittedSlotsPending() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ executor.setUseTimeoutCollection(true);
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ ((Configuration) context.getConfig())
+ .set(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS, 100L);
+ ((Configuration) context.getConfig()).set(AgentExecutionOptions.TOOL_CALL_PARALLELISM, 2);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>(
+ "batch-1",
+ String.class,
+ () -> {
+ Thread.sleep(200);
+ return "one";
+ });
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2",
+ String.class,
+ () -> {
+ Thread.sleep(200);
+ return "two";
+ });
+ TestDurableCallable<String> third =
+ new TestDurableCallable<>("batch-3", String.class, () -> "three");
+ TestDurableCallable<String> fourth =
+ new TestDurableCallable<>("batch-4", String.class, () -> "four");
+
+ List<Outcome<String>> outcomes =
+ context.durableExecuteAllAsync(List.of(first, second, third, fourth));
+
+ assertTrue(outcomes.get(0).isFailure());
+ assertTrue(outcomes.get(1).isFailure());
+ assertTrue(outcomes.get(2).isFailure());
+ assertTrue(outcomes.get(3).isFailure());
+ List<CallResult> persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults();
+ assertTrue(persisted.get(0).isFailure());
+ assertTrue(persisted.get(1).isFailure());
+ assertTrue(persisted.get(2).isPending());
+ assertTrue(persisted.get(3).isPending());
+ assertEquals(4, context.getDurableExecutionContext().getCurrentCallIndex());
+ executor.close();
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncTimeoutLeavesQueuedButUnstartedSlotsPending() throws Exception {
Review Comment:
This drives `InspectingContinuationActionExecutor.executeAllAsyncWithDeadline` at `:639`, which carries its own copy of `started.set(index, 1)` at `:670`. So it pins the double rather than `runtime/src/main/java21/.../ContinuationActionExecutor.java:217`, and deleting that line leaves every Java test green. There is no `src/test/java21` root, and the one e2e that does drive production java21, `AsyncExecutionTest.testToolCallBatchTimeoutKeepsCompletedOutcomes`, dispatches two calls at `tool-call.parallelism = 2` against the default `num-async-threads`, so nothing ever queues there. The Python side does cover the real path, at `test_flink_runner_context_reconcilable.py:899`.
`asyncExecutor` is `newFixedThreadPool(numAsyncThreads)`, so lowering `num-async-threads` under the batch parallelism in that e2e would queue a slot. Is that worth adding?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +293,158 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ parallelism = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_PARALLELISM)
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms > 0 else None
+ suppliers = [supplier for _, supplier in plan.suppliers]
+ batch_futures: list[Any | None] = [None] * len(suppliers)
+ started: list[bool] = [False] * len(suppliers)
+ try:
+ executed = yield from _execute_sliding_window_batch(
+ self._ctx.executor,
+ suppliers,
+ parallelism,
+ deadline,
+ timeout_ms,
+ batch_futures,
+ started,
+ )
+ except _BatchTimeoutError as exception:
+ executed = _collect_sliding_window_outcomes_on_timeout(
+ batch_futures, started, exception
+ )
+ return self._ctx._finalize_batch_execution(
+ self._calls, plan, started, executed
+ )
+
+
+class _BatchTimeoutError(TimeoutError):
+ """Raised when a durable batch exceeds its deadline."""
+
+
+def _mark_started_on_run(
+ supplier: Callable[[], Any], started: list[bool], index: int
+) -> Callable[[], Any]:
+ """Wrap a supplier so ``started[index]`` flips only when the worker truly runs.
+
+ A task queued in a saturated pool but cancelled before it executes keeps
+ ``started[index] == False``, so it is treated as never-run and stays
+ re-executable on recovery instead of being recorded as a timeout failure.
+ """
+
+ def _run() -> Any:
+ started[index] = True
+ return supplier()
+
+ return _run
+
+
+def _execute_sliding_window_batch(
+ executor: ThreadPoolExecutor,
+ suppliers: list[Any],
+ parallelism: int,
+ deadline: float | None,
+ timeout_ms: int,
+ futures: list[Any | None],
+ started: list[bool],
+) -> Any:
+ batch_size = len(suppliers)
+ if batch_size == 0:
+ return []
+
+ parallelism_limit = min(max(parallelism, 1), batch_size)
+ next_to_submit = 0
+ completed = 0
+ counted = [False] * batch_size
+
+ def in_flight() -> int:
+ return sum(
+ 1
+ for i in range(next_to_submit)
+ if futures[i] is not None and not futures[i].done()
+ )
+
+ while completed < batch_size:
+ if deadline is not None and time.monotonic() >= deadline:
+ timeout_message = (
+ f"Async durable batch execution timed out after {timeout_ms} ms"
+ )
+ raise _BatchTimeoutError(timeout_message)
+
+ while next_to_submit < batch_size and in_flight() < parallelism_limit:
+ index = next_to_submit
+ futures[index] = executor.submit(
+ _mark_started_on_run(suppliers[index], started, index)
+ )
+ next_to_submit += 1
+
+ for i in range(next_to_submit):
+ if not counted[i] and futures[i].done():
+ counted[i] = True
+ completed += 1
+
+ if completed < batch_size:
+ yield
+
+ return _collect_outcomes(futures)
+
+
+def _collect_sliding_window_outcomes_on_timeout(
+ futures: list[Any | None],
+ started: list[bool],
+ timeout_exception: BaseException,
+) -> list[Outcome]:
+ outcomes = []
+ for is_started, future in zip(started, futures, strict=True):
+ if not is_started or future is None:
Review Comment:
The `not is_started` short-circuit returns before the `future.cancel()` on `:410`, so a queued-but-unstarted slot is no longer cancelled. At `7e815996` those slots had `submitted[i] == True` and did reach `cancel()`, and on a `ThreadPoolExecutor` cancelling a work item that has not started yet returns `True` and the worker skips it, so the tool genuinely never ran. At head it runs: the batch returns, the slot correctly stays pending, the pool executes the tool a moment later and throws the result away, and recovery executes it again.
Java cannot do the same here (`CompletableFuture.cancel` will not pull a supplier back out of the executor), but Python can. Would keeping only the `future is None` guard work? `started[i]` still governs finalization at `:1033`, so the slot stays pending either way, and a cancelled future falls through to the timeout failure at `:417`. It also means `cancel()`'s return value rather than the flag decides whether it ran.
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -241,6 +294,136 @@ def __await__(self) -> Any:
return result
+class _DurableBatchAsyncExecutionResult(AsyncExecutionResult):
+ def __init__(self, ctx: "FlinkRunnerContext", calls: list[DurableCall]) -> None:
+ self._ctx = ctx
+ self._calls = calls
+
+ def __await__(self) -> Any:
+ plan = self._ctx._prepare_batch_execution(self._calls)
+ parallelism = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_PARALLELISM)
+ timeout_ms = self._ctx.config.get(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS)
+ deadline = time.monotonic() + timeout_ms / 1000 if timeout_ms > 0 else None
+ suppliers = [supplier for _, supplier in plan.suppliers]
+ batch_futures: list[Any | None] = [None] * len(suppliers)
+ try:
+ executed = yield from _execute_sliding_window_batch(
+ self._ctx.executor,
+ suppliers,
+ parallelism,
+ deadline,
+ timeout_ms,
+ batch_futures,
+ plan.submitted,
+ )
+ except _BatchTimeoutError as exception:
+ executed = _collect_sliding_window_outcomes_on_timeout(
+ batch_futures, plan.submitted, exception
+ )
+ return self._ctx._finalize_batch_execution(self._calls, plan, executed)
+
+
+class _BatchTimeoutError(TimeoutError):
+ """Raised when a durable batch exceeds its deadline."""
+
+
+def _execute_sliding_window_batch(
+ executor: ThreadPoolExecutor,
+ suppliers: list[Any],
+ parallelism: int,
+ deadline: float | None,
+ timeout_ms: int,
+ futures: list[Any | None],
+ submitted: list[bool],
+) -> Any:
+ batch_size = len(suppliers)
+ if batch_size == 0:
+ return []
+
+ parallelism_limit = min(max(parallelism, 1), batch_size)
+ next_to_submit = 0
+ completed = 0
+ counted = [False] * batch_size
+
+ def in_flight() -> int:
+ return sum(
+ 1
+ for i in range(next_to_submit)
+ if futures[i] is not None and not futures[i].done()
+ )
+
+ while completed < batch_size:
+ if deadline is not None and time.monotonic() >= deadline:
+ timeout_message = (
+ f"Async durable batch execution timed out after {timeout_ms} ms"
+ )
+ raise _BatchTimeoutError(timeout_message)
+
+ while next_to_submit < batch_size and in_flight() < parallelism_limit:
+ futures[next_to_submit] = executor.submit(suppliers[next_to_submit])
+ submitted[next_to_submit] = True
+ next_to_submit += 1
+
+ for i in range(next_to_submit):
+ if not counted[i] and futures[i].done():
+ counted[i] = True
+ completed += 1
+
+ if completed < batch_size:
+ yield
+
+ return _collect_outcomes(futures)
+
+
+def _collect_sliding_window_outcomes_on_timeout(
+ futures: list[Any | None],
+ submitted: list[bool],
+ timeout_exception: BaseException,
+) -> list[Outcome]:
+ outcomes = []
+ for is_submitted, future in zip(submitted, futures, strict=True):
+ if not is_submitted or future is None:
+ outcomes.append(Outcome.failure(timeout_exception))
+ continue
+ if not future.done():
+ future.cancel()
+ if future.done() and not future.cancelled():
Review Comment:
Closed from my side. Started semantics on both sides, traced through finalization on Java and Python. One consequence of the switch raised as a new comment on `_collect_sliding_window_outcomes_on_timeout`.
##########
docs/content/docs/operations/configuration.md:
##########
@@ -131,9 +131,11 @@ Here is the list of all built-in core configuration options.
| `max-retries` | 3 | int | Number of retries when using `ErrorHandlingStrategy.RETRY`. |
| `retry-wait-interval` | 1 | int | Base wait interval in seconds between retries when using `ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the connection name. |
| `chat.async` | true | boolean | Whether chat asynchronously for built-in chat action. |
-| `tool-call.async` | true | boolean | Whether process tool call for built-in tool call action. |
+| `tool-call.async` | true | boolean | Whether the built-in tool-call action runs each tool via durable async execution. |
+| `tool-call.parallelism` | os cpu count | int | In-flight concurrency for tool calls from one `ToolRequestEvent` batch when `tool-call.async` is enabled. `1` runs tools serially; values greater than `1` run a parallel durable batch with a sliding window of at most that many concurrent tool calls. On **Java**, concurrent in-batch execution requires **JDK 21+** (Continuation API); below JDK 21 the batch still runs but tool calls execute serially. **Python** uses the shared async `ThreadPoolExecutor` and runs batches concurrently regardless of JDK version. Increases in-flight external calls; after failover, unfinished tools may be submitted again — side-effecting tools should be idempotent or provide a reconciler. {{< hint warning >}}**Default is parallel** (`os cpu count`). Chat, RAG, and tool batches share one `num-async-threads` pool **per operator subtask** (all keys on that subtask). Built-in actions for a single key run one at a time, so chat
and a tool batch on the **same key** do not overlap in the usual chat → tool path; delay shows up mainly **across keys** on the same subtask. With defaults (`num-async-threads = 2× cores`, `tool-call.parallelism = cores`), one batch can use up to half the pool; several busy keys can still saturate it. Lower this value or increase `num-async-threads` on hot subtasks. {{< /hint >}} |
+| `tool-call.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall timeout for one parallel tool-call batch. Non-positive disables it. On timeout, completed slots keep their outcome and unfinished slots fail. Timeout cancellation is best-effort; external side effects from unfinished tool calls may still complete, so side-effecting tools should be idempotent or provide a reconciler. On **Java**, only enforced on **JDK 21+**; on JDK 11 the batch fallback ignores this setting and runs to completion serially. **Python** enforces the deadline in the batch await loop. |
Review Comment:
All three sites read correctly now, and the thread-reclamation note reached both doc pages. Closed.
##########
python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py:
##########
@@ -415,3 +594,458 @@ def collect_kwargs(**kwargs: Any) -> dict[str, Any]:
_close_runner_context(ctx)
assert result == {}
+
+
+def test_flink_runner_context_durable_execute_all_async_runs_calls_in_parallel() -> None:
+ j_runner_context = _FakeJavaRunnerContext()
+ config = AgentConfiguration(
+ {"tool-call.batch.timeout.ms": -1, "tool-call.parallelism": 3}
+ )
+ ctx = _create_runner_context(j_runner_context, config=config, executor_workers=3)
+ sleep_seconds = 0.2
+
+ def slow_call(value: str) -> str:
+ time.sleep(sleep_seconds)
+ return value
+
+ try:
+ start = time.perf_counter()
+ outcomes = _run_async(
+ ctx.durable_execute_all_async(
+ [
+ _durable_call(slow_call, "one"),
+ _durable_call(slow_call, "two"),
+ _durable_call(slow_call, "three"),
+ ]
+ )
+ )
+ elapsed = time.perf_counter() - start
+ finally:
+ _close_runner_context(ctx)
+
+ assert [outcome.value for outcome in outcomes] == ["one", "two", "three"]
+ assert elapsed < sleep_seconds * 2.5
Review Comment:
The barrier version is deterministic where the margin was not. Closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3792071175
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
Review Comment:
Good catch. There was a deadlock here, which is exactly why the CI below timed out. I've already fixed it in a new commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3793394857
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
Review Comment:
Checked this at head, including the timeout path. Closed from my side.
##########
python/flink_agents/api/runner_context.py:
##########
@@ -24,12 +25,49 @@
from flink_agents.api.metric_group import MetricGroup
from flink_agents.api.resource import Resource, ResourceType
-__all__ = ["AsyncExecutionResult", "RunnerContext"]
+__all__ = ["AsyncExecutionResult", "DurableCall", "Outcome", "RunnerContext"]
if TYPE_CHECKING:
from flink_agents.api.memory_object import MemoryObject
+@dataclass(frozen=True)
+class DurableCall:
+ """A deterministic durable call entry for batch execution."""
+
+ id: str
Review Comment:
Thanks for opening #1016, that covers it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3792072589
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java:
##########
@@ -477,7 +599,7 @@ protected <T> Optional<T> tryGetCachedResult(
} else if (resultPayload != null) {
return Optional.of(OBJECT_MAPPER.readValue(resultPayload, resultClass));
} else {
- return Optional.of(null);
+ return Optional.empty();
Review Comment:
This path will actually not be reached anymore. When switching from parallel to serial, `matchNextOrClearSubsequentCallResult` already returns `null` for slots that are in the pending state. However, to avoid introducing other potential errors, I will keep this null pointer exception as is; we can fix it later if any problems arise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3665328507
##########
runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java:
##########
@@ -240,12 +397,61 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) {
return supplier.get();
}
+ @Override
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ java.time.Duration timeout) {
+ executeAllAsyncCallCount++;
+ executeAllAsyncBatchSizes.add(suppliers.size());
+ if (useTimeoutCollection) {
+ return collectTimedOutOutcomes(suppliers);
+ }
+ List<Outcome<T>> outcomes = new java.util.ArrayList<>(suppliers.size());
+ for (Callable<T> supplier : suppliers) {
+ try {
+ outcomes.add(Outcome.success(supplier.call()));
+ } catch (Exception e) {
+ outcomes.add(Outcome.failure(e));
+ }
+ }
+ return outcomes;
+ }
+
+ private <T> List<Outcome<T>> collectTimedOutOutcomes(List<Callable<T>> suppliers) {
Review Comment:
I will add e2e tests to cover realistic timeout scenarios after the previous configuration issue is fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5141688476
hi @weiqingy ，Thank you for the very thorough review. I have addressed the timeout configuration type, naming, and the > 0 activation behavior; kept toolActionId consistent with the previous approach; and fixed the recovery issue when switching from parallel back to serial execution.
There is still one open point regarding internal exception handling in durableExecuteAllAsync.
For the exception boundary of `durableExecuteAllAsync`, I prefer a **two-layer model**: some failures become `Outcome`s (caller can decide how to handle them), while others must still propagate as thrown exceptions (infrastructure / invariants).
## Layer A: Should become `Outcome` (caller can decide)
| Case | Rationale |
|------|-----------|
| Single callable execution failure | Already `Outcome.failure`; `ToolCallAction` can record per-tool `ToolResponse` |
| Batch timeout for unfinished slots | Already `Outcome.failure(TimeoutException)`; partial success is meaningful |
| Per-slot result/exception serialization failure during finalize | Execution already finished; only persistence failed → that slot becomes failure, **without** affecting other successful slots’ `Outcome`s |
| Per-slot deserialization failure during replay | Treat as recovery failure for that slot; can fail and re-run without aborting the whole batch |
| Reconciler execution failure | Same as serial async: a per-call failure |
**Principle:** If the error is **per-call, expected, and the caller can turn it into a `ToolResponse` or log**, wrap it in `Outcome`.
## Layer B: Should still throw (caller must not swallow)
| Case | Rationale |
|------|-----------|
| `persistActionState()` failure (Fluss / state-backend IO) | Journal is untrustworthy; continuing may break durable semantics → fail fast and let Flink retry/recover |
| Illegal state machine in `finalizeCallAt` (no slot / not PENDING / identity mismatch) | Usually a bug or serious state corruption; `Outcome` would hide invariant violations |
| `reservePendingBatch` argument mismatch (ids/digests size) | Programming error; runtime should not swallow it |
| `Error` (e.g. OOM) | Must not be caught and converted to `Outcome` |
| Mailbox thread check failure | Breaks the concurrency model → must throw |
| Durable path taken without `durableExecutionContext` | Configuration / lifecycle bug |
I'm not sure this approach is beneficial for durableExecuteAllAsync — in that case, if ToolCallAction handles the exception here, it would mark all tool calls as failed. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5466986627
@da-daken Thanks for addressing all the comment! 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #926:
URL: https://github.com/apache/flink-agents/pull/926
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3864602380
##########
runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java:
##########
@@ -195,6 +208,357 @@ void testDurableExecuteAsyncReconcilableReconcileExceptionPersistsFailure() thro
assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
}
+ @Test
+ void testDurableExecuteAsyncCompletionOnlyReExecutesPendingSlot() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(CallResult.pending("tool-call", ""));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>("tool-call", String.class, () -> "recovered");
+
+ String result = context.durableExecuteAsync(callable);
+
+ assertEquals("recovered", result);
+ assertEquals(1, callable.getCallCount());
+ assertEquals(1, executor.getExecuteAsyncCallCount());
+ assertEquals(1, persistCallCount.get());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ CallResult persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults().get(0);
+ assertTrue(persisted.isSuccess());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncInitialBatchPersistsOutcomes() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>("batch-1", String.class, () -> "one");
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>("batch-2", String.class, () -> "two");
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(first, second));
+
+ assertEquals("one", outcomes.get(0).getValue());
+ assertEquals("two", outcomes.get(1).getValue());
+ assertEquals(1, executor.getExecuteAllAsyncCallCount());
+ assertEquals(List.of(2), executor.getExecuteAllAsyncBatchSizes());
+ assertEquals(1, first.getCallCount());
+ assertEquals(1, second.getCallCount());
+ assertEquals(3, persistCallCount.get());
+ assertEquals(2, context.getDurableExecutionContext().getCurrentCallIndex());
+ List<CallResult> persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults();
+ assertEquals(2, persisted.size());
+ assertEquals("batch-1", persisted.get(0).getFunctionId());
+ assertTrue(persisted.get(0).isSuccess());
+ assertEquals("batch-2", persisted.get(1).getFunctionId());
+ assertTrue(persisted.get(1).isSuccess());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncReconcilesPendingSlot() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(CallResult.pending("batch-1", ""));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestReconcilableCallable<String> callable =
+ new TestReconcilableCallable<>(
+ "batch-1",
+ String.class,
+ () -> fail("call should not be executed"),
+ () -> "recovered");
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertEquals("recovered", outcomes.get(0).getValue());
+ assertEquals(0, callable.getCallCount());
+ assertEquals(1, callable.getReconcileCount());
+ assertEquals(1, executor.getExecuteAllAsyncCallCount());
+ assertEquals(1, persistCallCount.get());
+ assertTrue(actionState.getCallResults().get(0).isSuccess());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncRecoversPartialFinalizedBatch() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(
+ 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", ""));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>(
+ "batch-1", String.class, () -> fail("cached slot should not execute"));
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2", String.class, () -> fail("cached slot should not execute"));
+ TestDurableCallable<String> third =
+ new TestDurableCallable<>("batch-3", String.class, () -> "fresh-three");
+
+ List<Outcome<String>> outcomes =
+ context.durableExecuteAllAsync(List.of(first, second, third));
+
+ assertEquals("cached-one", outcomes.get(0).getValue());
+ assertEquals("cached-two", outcomes.get(1).getValue());
+ assertEquals("fresh-three", outcomes.get(2).getValue());
+ assertEquals(0, first.getCallCount());
+ assertEquals(0, second.getCallCount());
+ assertEquals(1, third.getCallCount());
+ assertEquals(1, executor.getExecuteAllAsyncCallCount());
+ assertEquals(List.of(1), executor.getExecuteAllAsyncBatchSizes());
+ assertEquals("batch-3", actionState.getCallResults().get(2).getFunctionId());
+ assertTrue(actionState.getCallResults().get(2).isSuccess());
+ assertEquals(1, persistCallCount.get());
+ assertEquals(3, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncReturnsCachedFailureOutcome() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(
+ new CallResult(
+ "batch-1",
+ "",
+ null,
+ OBJECT_MAPPER.writeValueAsBytes(
+ RunnerContextImpl.DurableExecutionException.fromException(
+ new IllegalStateException("cached failure")))));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>(
+ "batch-1", String.class, () -> fail("cached slot should not execute"));
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertTrue(outcomes.get(0).isFailure());
+ assertInstanceOf(IllegalStateException.class, outcomes.get(0).getError());
+ assertTrue(outcomes.get(0).getError().getMessage().contains("cached failure"));
+ assertEquals(0, callable.getCallCount());
+ assertEquals(0, executor.getExecuteAllAsyncCallCount());
+ assertEquals(0, persistCallCount.get());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncReturnsDeserializeFailureAsOutcome() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ ActionState actionState = new ActionState(null);
+ actionState.addCallResult(
+ new CallResult(
+ "batch-1",
+ "",
+ "not-valid-json".getBytes(java.nio.charset.StandardCharsets.UTF_8),
+ null));
+ JavaRunnerContextImpl context = createContext(actionState, executor);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>(
+ "batch-1", String.class, () -> fail("cached slot should not execute"));
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertTrue(outcomes.get(0).isFailure());
+ assertInstanceOf(JsonProcessingException.class, outcomes.get(0).getError());
+ assertEquals(0, callable.getCallCount());
+ assertEquals(1, context.getDurableExecutionContext().getCurrentCallIndex());
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncPassesParallelismFromConfig() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ ((Configuration) context.getConfig()).set(AgentExecutionOptions.TOOL_CALL_PARALLELISM, 4);
+ TestDurableCallable<String> callable =
+ new TestDurableCallable<>("batch-1", String.class, () -> "ok");
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(callable));
+
+ assertEquals("ok", outcomes.get(0).getValue());
+ assertEquals(4, executor.getLastExecuteAllAsyncMaxParallelism());
+ executor.close();
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncTimeoutKeepsCompletedOutcomes() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ executor.setUseTimeoutCollection(true);
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ ((Configuration) context.getConfig())
+ .set(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS, 100L);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>("batch-1", String.class, () -> "fast");
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2",
+ String.class,
+ () -> {
+ Thread.sleep(200);
+ return "slow";
+ });
+
+ List<Outcome<String>> outcomes = context.durableExecuteAllAsync(List.of(first, second));
+
+ assertEquals("fast", outcomes.get(0).getValue());
+ assertTrue(outcomes.get(1).isFailure());
+ assertInstanceOf(TimeoutException.class, outcomes.get(1).getError());
+ assertEquals(Duration.ofMillis(100), executor.getLastExecuteAllAsyncTimeout());
+ assertEquals(1, first.getCallCount());
+ assertEquals(1, second.getCallCount());
+ List<CallResult> persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults();
+ assertTrue(persisted.get(0).isSuccess());
+ assertTrue(persisted.get(1).isFailure());
+ assertEquals(2, context.getDurableExecutionContext().getCurrentCallIndex());
+ executor.close();
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncTimeoutLeavesUnsubmittedSlotsPending() throws Exception {
+ InspectingContinuationActionExecutor executor = new InspectingContinuationActionExecutor();
+ executor.setUseTimeoutCollection(true);
+ JavaRunnerContextImpl context = createContext(new ActionState(null), executor);
+ ((Configuration) context.getConfig())
+ .set(AgentExecutionOptions.TOOL_CALL_BATCH_TIMEOUT_MS, 100L);
+ ((Configuration) context.getConfig()).set(AgentExecutionOptions.TOOL_CALL_PARALLELISM, 2);
+ TestDurableCallable<String> first =
+ new TestDurableCallable<>(
+ "batch-1",
+ String.class,
+ () -> {
+ Thread.sleep(200);
+ return "one";
+ });
+ TestDurableCallable<String> second =
+ new TestDurableCallable<>(
+ "batch-2",
+ String.class,
+ () -> {
+ Thread.sleep(200);
+ return "two";
+ });
+ TestDurableCallable<String> third =
+ new TestDurableCallable<>("batch-3", String.class, () -> "three");
+ TestDurableCallable<String> fourth =
+ new TestDurableCallable<>("batch-4", String.class, () -> "four");
+
+ List<Outcome<String>> outcomes =
+ context.durableExecuteAllAsync(List.of(first, second, third, fourth));
+
+ assertTrue(outcomes.get(0).isFailure());
+ assertTrue(outcomes.get(1).isFailure());
+ assertTrue(outcomes.get(2).isFailure());
+ assertTrue(outcomes.get(3).isFailure());
+ List<CallResult> persisted =
+ context.getDurableExecutionContext().getActionState().getCallResults();
+ assertTrue(persisted.get(0).isFailure());
+ assertTrue(persisted.get(1).isFailure());
+ assertTrue(persisted.get(2).isPending());
+ assertTrue(persisted.get(3).isPending());
+ assertEquals(4, context.getDurableExecutionContext().getCurrentCallIndex());
+ executor.close();
+ }
+
+ @Test
+ void testDurableExecuteAllAsyncTimeoutLeavesQueuedButUnstartedSlotsPending() throws Exception {
Review Comment:
Thanks! Covered by the new testToolCallBatchTimeoutCancelsQueuedButUnstartedSlots.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5327933802
@weiqingy Thanks for the review! The three latest issues are fixed. I've adopted the started semantics and unified the creation location for Java and Python—this was missed in the last commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3671217142
##########
api/src/main/java/org/apache/flink/agents/api/agents/AgentExecutionOptions.java:
##########
@@ -42,9 +44,43 @@ public class AgentExecutionOptions {
public static final ConfigOption<Boolean> CHAT_ASYNC =
new ConfigOption<>("chat.async", Boolean.class, true);
+ /** Whether the built-in tool-call action runs each tool via durable async execution. */
public static final ConfigOption<Boolean> TOOL_CALL_ASYNC =
new ConfigOption<>("tool-call.async", Boolean.class, true);
+ /**
+ * Whether multiple tool calls from one {@code ToolRequestEvent} run as one parallel durable
+ * batch when {@link #TOOL_CALL_ASYNC} is also enabled (JDK 21+).
+ *
+ * <p>Default is {@code true}. A parallel batch raises the number of in-flight external calls;
+ * after failover, tools whose results were not yet persisted may be submitted again.
+ * Side-effecting tools should be idempotent or provide a {@code reconciler()}. Set to {@code
+ * false} to keep serial async or sync tool execution.
+ */
+ public static final ConfigOption<Boolean> TOOL_CALL_PARALLEL =
+ new ConfigOption<>("tool-call.parallel", Boolean.class, true);
+
+ /**
+ * Size of the dedicated thread pool used for tool-call async and parallel batch execution.
+ *
+ * <p>Separate from {@link #NUM_ASYNC_THREADS} so a large tool batch does not exhaust the global
+ * async pool.
+ */
+ public static final ConfigOption<Integer> TOOL_CALL_NUM_ASYNC_THREADS =
+ new ConfigOption<>(
+ "tool-call.num-async-threads",
+ Integer.class,
+ Runtime.getRuntime().availableProcessors() * 2);
+
+ /**
+ * Overall timeout for one parallel tool-call batch.
+ *
+ * <p>Non-positive values disable the timeout. When the deadline elapses, unfinished slots are
+ * failed; slots that already completed keep their success or failure outcome.
+ */
+ public static final ConfigOption<Duration> TOOL_CALL_BATCH_TIMEOUT =
+ new ConfigOption<>("tool-call.batch.timeout", Duration.class, Duration.ofMillis(-1));
Review Comment:
The type change buys more than the YAML fix, for what it's worth — `AgentConfiguration.get()` has a `Long` branch at `AgentConfiguration.java:142`, and `Long` is Jackson-native, so the plan-serialization and Python routes resolve along with it.
Two smaller things while the option is in flux. `short-term-memory.state-ttl.ms` (`AgentExecutionOptions.java:88`) is the existing `Long`-millis option in this class and puts the unit in the key — since a bare `Long` drops the unit `Duration` carried, would `tool-call.batch.timeout.ms` be worth matching to it while the key is still unreleased?
And a few `Duration` leftovers will probably want sweeping with the same change: the parity-harness entries added for it (`check_java_python_config_options_parity.py:45` and the `_java_duration_to_millis` branch at `:99`) are covered by `"java.lang.Long": int` at `:40` once the type moves, and the config table still lists the type as `Duration` with a `-1ms` default (`configuration.md:137`). Any reason to keep those around once the type moves?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,6 +102,159 @@ private <T> T durableExecuteAsyncWithReconcile(
callable, reconcileCallable, () -> executeAsyncCallable(callable));
}
+ @Override
+ public <T> List<Outcome<T>> durableExecuteAllAsync(List<DurableCallable<T>> callables)
+ throws Exception {
+ if (callables.isEmpty()) {
+ return List.of();
+ }
+ if (durableExecutionContext == null) {
+ return executeAllWithoutDurableState(callables);
+ }
+
+ String argsDigest = "";
+ int base = durableExecutionContext.getCurrentCallIndex();
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base, argsDigest);
+
+ reservePendingBatchIfNeeded(callables, argsDigest, plan);
+
+ List<Outcome<T>> executed = executeOutcomeSuppliers(plan.suppliers);
+ finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+
+ advanceCallIndexBy(callables.size());
+ return plan.outcomes;
+ }
+
+ private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
+ List<DurableCallable<T>> callables, int base, String argsDigest) throws Exception {
+ BatchExecutionPlan<T> plan = new BatchExecutionPlan<>(callables.size());
+ for (int i = 0; i < callables.size(); i++) {
+ DurableCallable<T> callable = callables.get(i);
+ CallResult current = getCallResultAt(base + i);
+ if (current == null) {
+ markReservationStart(plan, i);
+ addExecutableCall(plan, i, callable::call);
+ continue;
+ }
+ if (!current.matches(callable.getId(), argsDigest)) {
+ clearCallResultsFromAndPersist(base + i);
+ plan.needsReservation = true;
+ plan.executionStart = i;
+ addExecutableCall(plan, i, callable::call);
+ appendRemainingExecutions(callables, plan, i + 1);
+ break;
+ }
+ if (current.isPending()) {
+ Callable<T> reconcileCallable = callable.reconciler();
+ Callable<T> executionCallable =
+ reconcileCallable != null ? reconcileCallable : callable::call;
+ addExecutableCall(plan, i, executionCallable);
+ } else {
+ plan.outcomes.add(
+ readTerminalOutcomeAt(
+ base + i, callable.getId(), argsDigest, callable.getResultClass()));
+ }
+ }
+ return plan;
+ }
+
+ private <T> void markReservationStart(BatchExecutionPlan<T> plan, int callIndex) {
+ plan.needsReservation = true;
+ if (plan.executionStart < 0) {
+ plan.executionStart = callIndex;
+ }
+ }
+
+ private <T> void addExecutableCall(
+ BatchExecutionPlan<T> plan, int callIndex, Callable<T> executionCallable) {
+ plan.outcomes.add(null);
+ plan.suppliers.add(executionCallable);
+ plan.executableCallIndexes.add(callIndex);
+ }
+
+ private <T> void appendRemainingExecutions(
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan, int startIndex) {
+ for (int i = startIndex; i < callables.size(); i++) {
+ DurableCallable<T> remaining = callables.get(i);
+ addExecutableCall(plan, i, remaining::call);
+ }
+ }
+
+ private <T> void reservePendingBatchIfNeeded(
+ List<DurableCallable<T>> callables, String argsDigest, BatchExecutionPlan<T> plan) {
+ if (!plan.needsReservation) {
+ return;
+ }
+ List<String> ids = new ArrayList<>();
+ for (DurableCallable<T> callable :
+ callables.subList(plan.executionStart, callables.size())) {
+ ids.add(callable.getId());
+ }
+ reservePendingBatch(ids, argsDigest);
Review Comment:
That plan sounds right. One part I wasn't sure it reaches: the read side is status-blind independently of how the slot got written. `CallResult.matches` (`CallResult.java:160`) compares only `functionId` and `argsDigest`, so a reserved PENDING slot would still read as a hit at the match site even once `finalizeCallAt` owns the write side. Does the serial path want a status guard there too, or does your fix remove the route that reaches it with a PENDING slot in the first place?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3746345187
##########
python/flink_agents/api/runner_context.py:
##########
@@ -24,12 +25,49 @@
from flink_agents.api.metric_group import MetricGroup
from flink_agents.api.resource import Resource, ResourceType
-__all__ = ["AsyncExecutionResult", "RunnerContext"]
+__all__ = ["AsyncExecutionResult", "DurableCall", "Outcome", "RunnerContext"]
if TYPE_CHECKING:
from flink_agents.api.memory_object import MemoryObject
+@dataclass(frozen=True)
+class DurableCall:
+ """A deterministic durable call entry for batch execution."""
+
+ id: str
Review Comment:
The code side is resolved: `DurableCall.id` is gone, and `tool_call_action.py` no longer imports from `flink_agents.runtime`.
On the tech-debt follow-up that was suggested for bringing Python's durable identity back in line with Java's `getId()` contract, I could not find an issue tracking it. The closest is #956, which is scoped to the Tool result contract (`ToolResponse` vs `Any`) and does not touch recovery identity or the class names. Would it be worth opening one before this merges? Deferrals without an issue behind them tend to quietly disappear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3665311413
##########
api/src/main/java/org/apache/flink/agents/api/agents/AgentExecutionOptions.java:
##########
@@ -42,9 +44,43 @@ public class AgentExecutionOptions {
public static final ConfigOption<Boolean> CHAT_ASYNC =
new ConfigOption<>("chat.async", Boolean.class, true);
+ /** Whether the built-in tool-call action runs each tool via durable async execution. */
public static final ConfigOption<Boolean> TOOL_CALL_ASYNC =
new ConfigOption<>("tool-call.async", Boolean.class, true);
+ /**
+ * Whether multiple tool calls from one {@code ToolRequestEvent} run as one parallel durable
+ * batch when {@link #TOOL_CALL_ASYNC} is also enabled (JDK 21+).
+ *
+ * <p>Default is {@code true}. A parallel batch raises the number of in-flight external calls;
+ * after failover, tools whose results were not yet persisted may be submitted again.
+ * Side-effecting tools should be idempotent or provide a {@code reconciler()}. Set to {@code
+ * false} to keep serial async or sync tool execution.
+ */
+ public static final ConfigOption<Boolean> TOOL_CALL_PARALLEL =
+ new ConfigOption<>("tool-call.parallel", Boolean.class, true);
+
+ /**
+ * Size of the dedicated thread pool used for tool-call async and parallel batch execution.
+ *
+ * <p>Separate from {@link #NUM_ASYNC_THREADS} so a large tool batch does not exhaust the global
+ * async pool.
+ */
+ public static final ConfigOption<Integer> TOOL_CALL_NUM_ASYNC_THREADS =
+ new ConfigOption<>(
+ "tool-call.num-async-threads",
+ Integer.class,
+ Runtime.getRuntime().availableProcessors() * 2);
+
+ /**
+ * Overall timeout for one parallel tool-call batch.
+ *
+ * <p>Non-positive values disable the timeout. When the deadline elapses, unfinished slots are
+ * failed; slots that already completed keep their success or failure outcome.
+ */
+ public static final ConfigOption<Duration> TOOL_CALL_BATCH_TIMEOUT =
+ new ConfigOption<>("tool-call.batch.timeout", Duration.class, Duration.ofMillis(-1));
Review Comment:
Thanks for the catch. I missed the YAML case and have now updated the timeout configuration to use the `Long `type
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3665417925
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java:
##########
@@ -79,55 +103,160 @@ public static void processToolRequest(Event event, RunnerContext ctx) {
diagnosticError = e.getMessage();
}
- if (tool != null) {
- try {
- // Framework-owned injected args must win over model-provided values so hidden
- // context such as tenant ids cannot be spoofed by a tool call payload.
- mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
- ToolResponse response;
- final Tool toolRef = tool;
- final Map<String, Object> callArguments = mergedArguments;
- DurableCallable<ToolResponse> callable =
- new DurableCallable<>() {
- @Override
- public String getId() {
- return "tool-call";
- }
-
- @Override
- public Class<ToolResponse> getResultClass() {
- return ToolResponse.class;
- }
-
- @Override
- public ToolResponse call() throws Exception {
- return toolRef.call(new ToolParameters(callArguments));
- }
- };
- response =
- toolCallAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
- success.put(id, response.isSuccess());
- responses.put(id, response);
- if (!response.isSuccess() && response.getError() != null) {
- error.put(id, response.getError());
- }
- } catch (Exception e) {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s execute failed.", name)));
- error.put(id, e.getMessage());
- }
- } else {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s does not exist.", name)));
- error.put(id, diagnosticError != null ? diagnosticError : "Tool does not exist.");
+ if (tool == null) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s does not exist.", name)),
+ diagnosticError != null ? diagnosticError : "Tool does not exist.",
+ success,
+ error,
+ responses);
+ continue;
}
+
+ try {
+ // Framework-owned injected args must win over model-provided values so hidden
+ // context such as tenant ids cannot be spoofed by a tool call payload.
+ mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
+ } catch (Exception e) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s execute failed.", name)),
+ e.getMessage(),
+ success,
+ error,
+ responses);
+ continue;
+ }
+
+ final Tool toolRef = tool;
+ final Map<String, Object> callArguments = mergedArguments;
+ DurableCallable<ToolResponse> callable =
+ new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "tool-call-" + id;
+ }
+
+ @Override
+ public Class<ToolResponse> getResultClass() {
+ return ToolResponse.class;
+ }
+
+ @Override
+ public ToolResponse call() throws Exception {
+ return toolRef.call(new ToolParameters(callArguments));
+ }
+ };
+ executions.add(new ToolCallExecution(id, name, callable));
+ }
+ return executions;
+ }
+
+ private static void executeParallel(
+ List<ToolCallExecution> executions,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses) {
+ List<DurableCallable<ToolResponse>> callables = new ArrayList<>(executions.size());
+ for (ToolCallExecution execution : executions) {
+ callables.add(execution.callable);
+ }
+ try {
+ List<Outcome<ToolResponse>> outcomes = ctx.durableExecuteAllAsync(callables);
+ for (int i = 0; i < outcomes.size(); i++) {
+ recordOutcome(executions.get(i), outcomes.get(i), success, error, responses);
+ }
+ } catch (Exception e) {
Review Comment:
I'll handle all exceptions inside `durableExecuteAllAsync`, so the caller only needs to process the `outcome` without additional error handling. This keeps the outer logic cleaner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3792071970
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context, Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1), batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
+ context.setPendingBatchFuture(batchBarrier, deadlineNanos);
+
+ while (completed < batchSize) {
+ if (System.nanoTime() >= deadlineNanos) {
+ TimeoutException exception =
+ new TimeoutException(
+ "Async durable batch execution timed out after " + timeout);
+ batchBarrier.cancel(true);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomesOnTimeout(slots, exception);
+ }
+
+ while (nextToSubmit < batchSize && countInFlight(slots, nextToSubmit) < parallelismLimit) {
+ int index = nextToSubmit++;
+ Callable<T> supplier = suppliers.get(index);
+ slots[index] =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return Outcome.success(supplier.call());
+ } catch (Exception e) {
+ return Outcome.failure(e);
+ }
+ }, asyncExecutor);
+ }
+
+ for (int i = 0; i < nextToSubmit; i++) {
+ if (!counted[i] && slots[i].isDone()) {
+ counted[i] = true;
+ completed++;
+ }
+ }
+
+ if (completed < batchSize) {
+ Continuation.yield(SCOPE);
+ }
+ }
+
+ batchBarrier.complete(null);
+ context.setPendingBatchFuture(null);
+ return collectBatchOutcomes(Arrays.asList(slots));
+ }
+
+ private static <T> int countInFlight(
+ CompletableFuture<Outcome<T>>[] slots, int submittedCount) {
+ int inFlight = 0;
+ for (int i = 0; i < submittedCount; i++) {
+ if (!slots[i].isDone()) {
+ inFlight++;
+ }
+ }
+ return inFlight;
+ }
+
+
+ /**
+ * Collects per-slot outcomes after the batch barrier completes normally.
+ *
+ * <p>Each supplier already wraps success and failure into an {@link Outcome}, so {@code join()}
+ * returns that outcome rather than throwing for ordinary tool exceptions.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomes(
+ List<CompletableFuture<Outcome<T>>> futures) {
+ List<Outcome<T>> results = new ArrayList<>(futures.size());
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ results.add(future.join());
+ }
+ return results;
+ }
+
+ /**
+ * Collects per-slot outcomes when the batch deadline elapses.
+ *
+ * <p>Completed slots keep their success or failure outcome. Only slots that are still running
+ * (or become cancelled) are finalized as timeout failures. {@code cancel(true)} is attempted
+ * only for unfinished futures; a future that completes between the check and cancel stays
+ * non-cancelled and is collected as a normal outcome.
+ */
+ private static <T> List<Outcome<T>> collectBatchOutcomesOnTimeout(
+ CompletableFuture<Outcome<T>>[] futures, TimeoutException timeoutException) {
+ List<Outcome<T>> results = new ArrayList<>(futures.length);
+ for (CompletableFuture<Outcome<T>> future : futures) {
+ if (future == null) {
+ results.add(Outcome.failure(timeoutException));
Review Comment:
> Would leaving those slots PENDING, so recovery can still run them, fit better than recording them as failed?
Thanks for catching that – this looks friendlier than just failing everything. I'll go with your suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Parallel Tool Call Execution [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
da-daken commented on PR #926:
URL: https://github.com/apache/flink-agents/pull/926#issuecomment-5308147642
@weiqingy Thank you for your review. I am sorry for the delayed response. All the issues you raised have now been addressed in the latest commits. Please take another look when you have a moment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org