eemario commented on code in PR #1208:
URL: https://github.com/apache/flink-agents/pull/1208#discussion_r4228037899
##########
docs/content/docs/operations/configuration.md:
##########
@@ -134,10 +134,10 @@ Here is the list of all built-in core configuration
options.
| `retry-wait-interval` | 1 | int
| Base wait interval in seconds between retries. 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 configured ChatModel resource name. |
| `chat.async` | true | boolean
| Whether chat asynchronously for built-in chat 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; slots
that started but did not finish are recorded as failures; slots that never
started executing (for example, queued in a saturated pool) stay pending, so
they are re-executed after recovery instead of recording a false failure.
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. {{< hint warning >}}**Thread
reclamation:** a timeout unblocks the batch but cannot interrupt a tool that is
still running; its worker thread stays in the shared `num-async-threads` pool
until th
e tool returns on its own, so a tool that never returns permanently reduces
pool capacity. Bound blocking work inside the tool (for example an HTTP client
read timeout) rather than relying on this timeout to free the thread.{{< /hint
>}} |
+| `async.batch.parallelism` | os cpu count | int | Maximum concurrent calls in
one `ctx.gather(...)` batch, including ordinary calls, durable calls, and
built-in async tool calls. `1` runs calls sequentially. Actual concurrency is
also limited by `num-async-threads`; lower this per-batch limit to leave
capacity for other calls on the same subtask. Java and Python support batch
concurrency on all supported JDK versions. |
Review Comment:
Question on the API contract: config keys are part of the public API surface
— is a rename with no migration note the intended behavior here?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,35 +76,89 @@ protected <T> T resolveDurableAsync(DurableCallable<T>
callable) throws Exceptio
return durableExecuteAsyncWithReconcile(callable,
reconcileCallable);
}
}
- return durableExecuteCompletionOnly(callable, () ->
executeAsyncCallable(callable));
+ return durableExecuteCompletionOnly(callable, () ->
resolveAsync(callable::call));
}
private <T> T durableExecuteAsyncWithReconcile(
DurableCallable<T> callable, Callable<T> reconcileCallable) throws
Exception {
return durableExecuteWithReconcile(
- callable, reconcileCallable, () ->
executeAsyncCallable(callable));
+ callable, reconcileCallable, () ->
resolveAsync(callable::call));
}
@Override
- protected <T> List<Outcome<T>>
resolveDurableBatch(List<DurableCallable<T>> callables)
+ protected <T> List<Outcome<T>>
resolveAsyncBatch(List<SingleAsyncFuture<T>> futures)
throws Exception {
- if (callables.isEmpty()) {
- return List.of();
+ List<DurableCallable<T>> durableCalls = new ArrayList<>();
+ for (SingleAsyncFuture<T> future : futures) {
+ if (future.getDurableCallable() != null) {
+ durableCalls.add(future.getDurableCallable());
+ }
}
- if (durableExecutionContext == null) {
- return executeAllWithoutDurableState(callables);
+ if (durableExecutionContext == null || durableCalls.isEmpty()) {
+ List<Callable<T>> suppliers = new ArrayList<>();
+ for (SingleAsyncFuture<T> future : futures) {
+ suppliers.add(future.getCallable());
+ }
+ return executeAllWithoutDurableState(suppliers);
Review Comment:
This new ordinary path exposes an existing JDK 21 batch-executor bug.
When a callback throws Error, CompletableFuture.join() wraps it in
CompletionException. AsyncFutureImpl.await() then treats that wrapper as an
ordinary Exception, marks the handle done, and caches it.
Reproducer:
var batch = ctx.gather(List.of(ctx.executeAsync(() -> {
throw new AssertionError("fatal");
})));
batch.await();
On the real multi-release JDK 21 path, the first await throws
CompletionException(cause=AssertionError), and the second await rethrows the
same
cached wrapper. JDK < 21 instead propagates the original AssertionError
without
caching it.
This contradicts the PR's "Java fatal Errors propagate" contract and lets
catch (Exception) accidentally swallow a fatal Error. Could the JDK 21 normal
and timeout outcome collectors unwrap CompletionException and rethrow Error
directly, with a packaged multi-release JDK 21 regression test?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -222,25 +277,22 @@ private BatchExecutionPlan(int size) {
}
}
- private <T> List<Outcome<T>>
executeAllWithoutDurableState(List<DurableCallable<T>> callables)
+ private <T> List<Outcome<T>>
executeAllWithoutDurableState(List<Callable<T>> suppliers)
throws Exception {
- List<Callable<T>> suppliers = new ArrayList<>();
- for (DurableCallable<T> callable : callables) {
- suppliers.add(callable::call);
- }
List<Outcome<T>> outcomes =
executeOutcomeSuppliers(suppliers).getOutcomes();
for (Outcome<T> outcome : outcomes) {
- if (outcome.isFailure() && outcome.getError() instanceof
InterruptedException) {
- // Without a durable store there is nothing to leave pending,
but a cancellation
- // signal folded into a failed Outcome must still propagate as
an interruption
- // instead of being handed to the caller as an ordinary tool
failure.
- Thread.currentThread().interrupt();
- throw (InterruptedException) outcome.getError();
- }
+ rethrowInterruption(outcome);
}
return outcomes;
}
+ private static void rethrowInterruption(Outcome<?> outcome) throws
InterruptedException {
Review Comment:
This only recognizes a top-level InterruptedException. The Python runtime
and ModelRoutingResolver also classify ClosedByInterruptException and a
nested
InterruptedException as cancellation, while excluding bare
InterruptedIOException network timeouts.
With the new ordinary gather path, a callback that throws
ClosedByInterruptException is returned as Outcome.failure and cached on the
child handle; a later await rethrows the same cached exception instead of
retrying. The durable path can similarly persist it as a business failure.
Could AsyncFutureImpl.await(), this helper, and durable finalization share a
cause-chain interruption classifier, with direct, NIO, and
wrapped-interruption
tests?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
