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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
