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