This is an automated email from the ASF dual-hosted git repository.
yunfengzhou-hub pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 9af407ed8 [runtime][java][python] Avoid async-pool self-deadlock in
internal sub-agent await (#1213) (#1237)
9af407ed8 is described below
commit 9af407ed860e382a16434e46d49690adfe8b8d48
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Mon Oct 12 17:24:16 2026 +0800
[runtime][java][python] Avoid async-pool self-deadlock in internal
sub-agent await (#1213) (#1237)
With num-async-threads=1 (and under enough nested or concurrent waits at any
size), a parent action awaiting an internal sub-agent held the only pool
worker
via a blocking responseFuture.get(), while the child's own durable async
call
needed a worker from the same finite pool to make progress. The result was a
self-deadlock: neither side could advance.
The fix lets the parent's wait yield its continuation instead of occupying a
pool worker. A new runtime-internal marker interface,
ContinuationNativeCallable,
identifies callables whose body suspends by yielding the caller's
continuation
and must therefore run inline on the mailbox thread rather than being
dispatched
to an async-pool worker. JavaRunnerContextImpl.executionBoundary() routes
marked
callables through callable::call; all others continue to go through
resolveAsync.
resolveDurableAsync() now computes that boundary once and passes it to both
the
completion-only and the reconcile path, so a marked callable that also
supplies a
reconciler() cannot be dispatched to a pool worker either.
ContinuationActionExecutor
.awaitFuture() bridges the target CompletableFuture into a Future<?>
registered as
pendingFuture and yields via Continuation.yield, so executeAction's existing
hasPendingAsync gate drives resumption through the operator's mail loop.
Zero pool
workers are held during the wait, so deadlock is impossible at any pool
size,
nesting depth, or concurrency level. Durable state still records the real
SubagentResult because the callable returns it directly.
InternalSubagentSetup.prepare() now returns a ContinuationNativeCallable and
awaits the child via JavaRunnerContextImpl.awaitCompletableFuture. The
public
blocking RunnerContextImpl.awaitSubagentCall entry point used by the Python
cross-language path is preserved unchanged. DeferredSubagentFuture is
untouched;
the combined awaitAll path still routes through
DeferredSubagentFuture.execute()
and therefore through the marker.
Also fixes a pre-existing flake in the Python async batch timeout tests
surfaced
by CI on macos/py3.10. A batch deadline is wall-clock from the first
submission,
so asserting that a slot was started when the timeout hits left thread
startup
racing that deadline; under load the worker had not begun and the slot was
persisted as PENDING instead of FAILED. The tests now drive the batch one
step,
wait for their workers to announce they began, and only then resume, which
makes
"started but unfinished" hold whatever the startup latency. Shrinking the
timeout
to 1ms reproduces the original failure and still passes with the fix.
Three @Timeout(30) tests at num-async-threads=1 cover the issue's full
stated
scope (single call, nested calls, combined concurrent calls). All three
time out
without the fix and complete promptly with it.
JavaRunnerContextImplDurableExecuteAsyncTest
gains a case pinning that a marked callable supplying a reconciler() still
runs
inline. Full runtime suites pass on both JDK 11 and JDK 21.
AI-Contributed/Feature: 35/196
AI-Contributed/UT: 142/375
Co-authored-by: Qoder <[email protected]>
---
.../test_flink_runner_context_reconcilable.py | 101 ++++++---
.../runtime/async/ContinuationActionExecutor.java | 11 +
.../runtime/async/ContinuationNativeCallable.java | 34 +++
.../runtime/context/JavaRunnerContextImpl.java | 35 +++-
.../subagent/BaseDeferredSubagentSetup.java | 6 +-
.../runtime/subagent/InternalSubagentSetup.java | 72 +++++--
.../runtime/async/ContinuationActionExecutor.java | 38 ++++
...vaRunnerContextImplDurableExecuteAsyncTest.java | 23 ++
.../runtime/context/TestDurableCallables.java | 18 ++
.../subagent/InternalSubagentDeadlockTest.java | 233 +++++++++++++++++++++
10 files changed, 516 insertions(+), 55 deletions(-)
diff --git
a/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
b/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
index 92c0e154a..f0456d870 100644
---
a/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
+++
b/python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py
@@ -246,17 +246,47 @@ def
test_chat_failure_outcome_replays_without_provider_call(asynchronous: bool)
assert store.call_results[0].exception_payload is None
-def _run_async(result: Any) -> object:
- iterator = result.__await__()
- value = None
+def _drain(iterator: Any) -> object:
while True:
try:
- iterator.send(value)
- value = None
+ iterator.send(None)
except StopIteration as e: # noqa: PERF203
return e.value
+def _run_async(result: Any) -> object:
+ return _drain(result.__await__())
+
+
+_WORKER_START_TIMEOUT_SECONDS = 30.0
+
+
+def _await_workers(started: list[threading.Event]) -> None:
+ deadline = time.monotonic() + _WORKER_START_TIMEOUT_SECONDS
+ for index, event in enumerate(started):
+ remaining = max(deadline - time.monotonic(), 0.0)
+ assert event.wait(remaining), f"async worker {index} never started"
+
+
+def _run_async_after_workers_started(
+ result: Any, started: list[threading.Event]
+) -> object:
+ """Run a batch, pausing after submission until its workers have begun.
+
+ A batch deadline is wall-clock from the first submission, so a test
asserting
+ that a slot was *started* when the timeout hits cannot leave thread startup
+ racing that deadline: under CI load the worker has not begun yet and the
slot
+ is persisted as PENDING instead of FAILED. One generator step submits the
+ batch; waiting there until every worker reports it began makes "started but
+ unfinished" hold by construction, and the deadline has certainly passed by
the
+ time driving resumes.
+ """
+ iterator = result.__await__()
+ iterator.send(None)
+ _await_workers(started)
+ return _drain(iterator)
+
+
def _preload_pending(
j_runner_context: _FakeJavaRunnerContext,
func: Callable[..., Any],
@@ -1052,27 +1082,32 @@ def
test_flink_runner_context_gather_collects_failures() -> None:
def test_flink_runner_context_gather_timeout_keeps_completed_results() -> None:
j_runner_context = _FakeJavaRunnerContext()
- config = AgentConfiguration({"async.batch.timeout.ms": 100})
+ config = AgentConfiguration(
+ {"async.batch.timeout.ms": 100, "async.batch.parallelism": 2}
+ )
ctx = _create_runner_context(j_runner_context, config=config)
# The slow worker blocks on an event instead of racing a sleep against the
- # deadline. Once its worker thread starts, _mark_started_on_run has already
- # flipped the started flag, so the timeout must record it as FAILED
(started
- # but unfinished) rather than PENDING. The event guarantees it never
- # finishes before the deadline, and the finally block releases it so the
- # pool can shut down.
+ # deadline, and announces itself so the driver can wait for it. Once its
worker
+ # thread starts, _mark_started_on_run has already flipped the started
flag, so
+ # the timeout must record it as FAILED (started but unfinished) rather than
+ # PENDING. The event guarantees it never finishes before the deadline, and
the
+ # finally block releases it so the pool can shut down.
release = threading.Event()
+ slow_started = threading.Event()
def slow_call() -> str:
+ slow_started.set()
release.wait(5)
return "slow"
try:
- outcomes = _run_async(
+ outcomes = _run_async_after_workers_started(
ctx.gather(
ctx.durable_execute_async(lambda: "fast"),
ctx.durable_execute_async(slow_call),
- )
+ ),
+ [slow_started],
)
finally:
release.set()
@@ -1096,22 +1131,26 @@ def
test_flink_runner_context_gather_timeout_leaves_unsubmitted_slots_pending()
# The two blocking calls saturate the parallelism budget and hold their
# worker threads (started flag flipped) until the deadline, so they fail;
the
# remaining two suppliers are never submitted and stay pending. Blocking
on an
- # event instead of sleeping makes "started" deterministic rather than
racing
- # thread startup against a short wall-clock timeout.
+ # event instead of sleeping, and waiting for both workers to announce
+ # themselves, makes "started" deterministic rather than racing thread
startup
+ # against a short wall-clock timeout.
release = threading.Event()
+ started_events = {"one": threading.Event(), "two": threading.Event()}
def blocking_call(value: str) -> str:
+ started_events[value].set()
release.wait(5)
return value
try:
- outcomes = _run_async(
+ outcomes = _run_async_after_workers_started(
ctx.gather(
ctx.durable_execute_async(blocking_call, "one"),
ctx.durable_execute_async(blocking_call, "two"),
ctx.durable_execute_async(_call_value, "three"),
ctx.durable_execute_async(_call_value, "four"),
- )
+ ),
+ [started_events["one"], started_events["two"]],
)
finally:
release.set()
@@ -1141,12 +1180,15 @@ def
test_flink_runner_context_gather_timeout_leaves_queued_slots_pending() -> No
# Blocking on an event (rather than sleeping) keeps the two running workers
# deterministically "started but unfinished" until the deadline, while the
two
- # queued suppliers never begin.
+ # queued suppliers never begin. Waiting for both workers to announce
themselves
+ # removes the race between thread startup and the wall-clock deadline.
release = threading.Event()
+ started_events = {"one": threading.Event(), "two": threading.Event()}
executions: list[str] = []
def blocking_call(value: str) -> str:
executions.append(value)
+ started_events[value].set()
release.wait(5)
return value
@@ -1155,13 +1197,14 @@ def
test_flink_runner_context_gather_timeout_leaves_queued_slots_pending() -> No
return value
try:
- outcomes = _run_async(
+ outcomes = _run_async_after_workers_started(
ctx.gather(
ctx.durable_execute_async(blocking_call, "one"),
ctx.durable_execute_async(blocking_call, "two"),
ctx.durable_execute_async(queued_call, "three"),
ctx.durable_execute_async(queued_call, "four"),
- )
+ ),
+ [started_events["one"], started_events["two"]],
)
finally:
release.set()
@@ -1532,15 +1575,19 @@ def
test_mixed_batch_timeout_preserves_unstarted_durable_slot_for_recovery() ->
try:
# The cached durable child takes no worker; the ordinary child
occupies the
- # only worker, leaving the new durable child queued until the timeout.
+ # only worker, leaving the new durable child queued until the timeout.
The
+ # driver waits for that worker to announce itself, so "the only worker
is
+ # occupied when the deadline trips" does not race thread startup.
assert ctx.durable_execute(lambda: "cached", durable_id="cached") ==
"cached"
j_ctx.current_call_index = 0
- outcomes = _run_async(ctx.gather(
- ctx.durable_execute_async(lambda: "cached", durable_id="cached"),
- ctx.execute_async(ordinary),
- ctx.durable_execute_async(durable, durable_id="pending"),
- ))
- assert started.is_set()
+ outcomes = _run_async_after_workers_started(
+ ctx.gather(
+ ctx.durable_execute_async(lambda: "cached",
durable_id="cached"),
+ ctx.execute_async(ordinary),
+ ctx.durable_execute_async(durable, durable_id="pending"),
+ ),
+ [started],
+ )
assert outcomes[0].value == "cached"
assert isinstance(outcomes[1].error, TimeoutError)
assert isinstance(outcomes[2].error, TimeoutError)
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
index 37769a072..6132ae039 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
@@ -25,6 +25,7 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
/**
@@ -136,6 +137,16 @@ public class ContinuationActionExecutor {
}
}
+ /**
+ * Awaits a {@link CompletableFuture} by blocking the calling thread. On
JDK 21+ the
+ * multi-release variant yields the continuation instead, avoiding
async-pool exhaustion (issue
+ * #1213).
+ */
+ public <T> T awaitFuture(ContinuationContext context, CompletableFuture<T>
future)
+ throws Exception {
+ return future.get();
+ }
+
/** Stops async work without waiting for worker exit cleanup. */
public void shutdown() {
batchExecutor.shutdown();
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationNativeCallable.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationNativeCallable.java
new file mode 100644
index 000000000..d9bd89e36
--- /dev/null
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/async/ContinuationNativeCallable.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.async;
+
+import org.apache.flink.agents.api.context.DurableCallable;
+
+/**
+ * Marks a {@link DurableCallable} that suspends by yielding the caller's
continuation, and must
+ * therefore run inline on the mailbox thread instead of being dispatched to
an async-pool worker.
+ *
+ * <p>A waiting caller that occupies a pool worker holds a resource its own
callee may need to make
+ * progress, which self-deadlocks once concurrent or nested waits exhaust the
pool (issue #1213). A
+ * callable marked here awaits by yielding, so the worker pool stays free for
the callee.
+ *
+ * <p>Only meaningful on JDK 21+, where the continuation the callable yields
is the caller's; the
+ * callers that produce such callables gate on {@code
requireMailboxSuspension} beforehand.
+ */
+public interface ContinuationNativeCallable<T> extends DurableCallable<T> {}
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
index 27b402969..d4f541637 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java
@@ -27,6 +27,7 @@ import org.apache.flink.agents.runtime.actionstate.CallResult;
import org.apache.flink.agents.runtime.async.BatchExecutionResult;
import org.apache.flink.agents.runtime.async.ContinuationActionExecutor;
import org.apache.flink.agents.runtime.async.ContinuationContext;
+import org.apache.flink.agents.runtime.async.ContinuationNativeCallable;
import org.apache.flink.agents.runtime.metrics.FlinkAgentsMetricGroupImpl;
import java.time.Duration;
@@ -35,6 +36,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -71,19 +73,29 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
@Override
protected <T> T resolveDurableAsync(DurableCallable<T> callable) throws
Exception {
+ Callable<T> boundary = executionBoundary(callable);
if (durableExecutionContext != null) {
Callable<T> reconcileCallable = callable.reconciler();
if (reconcileCallable != null) {
- return durableExecuteAsyncWithReconcile(callable,
reconcileCallable);
+ return durableExecuteWithReconcile(callable,
reconcileCallable, boundary);
}
}
- return durableExecuteCompletionOnly(callable, () ->
resolveAsync(callable::call));
+ return durableExecuteCompletionOnly(callable, boundary);
}
- private <T> T durableExecuteAsyncWithReconcile(
- DurableCallable<T> callable, Callable<T> reconcileCallable) throws
Exception {
- return durableExecuteWithReconcile(
- callable, reconcileCallable, () ->
resolveAsync(callable::call));
+ /**
+ * Chooses where one durable async call executes, on both the
completion-only and the reconcile
+ * path. A {@link ContinuationNativeCallable} suspends by yielding the
caller's continuation, so
+ * it runs inline on the mailbox thread; handing it to an async-pool
worker would make the
+ * waiting caller hold a worker its own callee needs to progress (issue
#1213), and the yield
+ * itself would fail off the mailbox thread. Any other callable is
offloaded to the pool, which
+ * lets the continuation yield while it runs.
+ */
+ private <T> Callable<T> executionBoundary(DurableCallable<T> callable) {
+ if (callable instanceof ContinuationNativeCallable) {
+ return callable::call;
+ }
+ return () -> resolveAsync(callable::call);
}
@Override
@@ -350,6 +362,17 @@ public class JavaRunnerContextImpl extends
RunnerContextImpl {
}
}
+ /**
+ * Awaits a {@link CompletableFuture} by yielding the continuation,
without occupying an
+ * async-pool worker. Falls back to blocking {@code get()} when
continuations are unavailable.
+ */
+ public <T> T awaitCompletableFuture(CompletableFuture<T> future) throws
Exception {
+ if (continuationExecutor != null && continuationContext != null) {
+ return continuationExecutor.awaitFuture(continuationContext,
future);
+ }
+ return future.get();
+ }
+
@Override
public void close() throws Exception {
super.close();
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseDeferredSubagentSetup.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseDeferredSubagentSetup.java
index 56b7cedf8..7d840cd94 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseDeferredSubagentSetup.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseDeferredSubagentSetup.java
@@ -58,8 +58,10 @@ public abstract class BaseDeferredSubagentSetup extends
BaseSubagentSetup {
*
* <p>Called exactly once per invocation, when the deferred handle is
first resolved, on the
* mailbox thread. Implementations may therefore perform the
mailbox-confined part of issuing
- * the request here; the returned callable's {@link
DurableCallable#call()} carries only the
- * part that runs off the mailbox thread.
+ * the request here; the returned callable's {@link
DurableCallable#call()} then carries the
+ * wait, normally on an async-pool worker off the mailbox thread. A
callable marked {@link
+ * org.apache.flink.agents.runtime.async.ContinuationNativeCallable}
instead runs inline on the
+ * mailbox thread and releases it by suspending the caller's continuation.
*
* <p>The callable folds its own comprehensible failures into the returned
{@link
* SubagentResult}; an exception escaping {@link DurableCallable#call()}
is a system-level
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
index 9a660fe90..774205210 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
@@ -33,6 +33,7 @@ import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.agents.runtime.ResourceCache;
import org.apache.flink.agents.runtime.async.ContinuationActionExecutor;
+import org.apache.flink.agents.runtime.async.ContinuationNativeCallable;
import org.apache.flink.agents.runtime.condition.ActionMatcher;
import org.apache.flink.agents.runtime.context.JavaRunnerContextImpl;
import org.apache.flink.agents.runtime.context.RunnerContextImpl;
@@ -54,8 +55,10 @@ import java.util.Map;
* <p>Execution mode is the deferred one, inherited from {@link
BaseDeferredSubagentSetup}: {@link
* #submit} returns a deferred handle whose request is prepared on first
resolve. {@link #prepare}
* then runs the mailbox-confined bootstrap — registering the call status and
sending one {@code
- * InternalSubagentCallEvent} — and returns the durable callable that waits
off the mailbox thread
- * for the child to quiesce, so the operator can dispatch the child agent's
actions in between.
+ * InternalSubagentCallEvent} — and returns the durable callable that waits
for the child to
+ * quiesce. Unlike the other deferred setups, that callable runs inline on the
mailbox thread and
+ * releases it by yielding the caller's continuation, so the operator can
dispatch the child agent's
+ * actions in between while no async-pool worker is held.
*
* <p>Orchestration state lives here, not in the operator: the per-call
quiesce statuses, the
* per-scope child resource caches, and the per-key session index used to
clean up when a record
@@ -158,8 +161,15 @@ public class InternalSubagentSetup extends
BaseDeferredSubagentSetup {
}
/**
- * The mailbox-confined bootstrap plus the off-mailbox wait. Sending the
bootstrap event must
- * happen before the wait releases the mailbox, so it runs here, on the
resolving thread.
+ * The mailbox-confined bootstrap plus the wait. Sending the bootstrap
event must happen before
+ * the wait releases the mailbox, so it runs here, on the resolving thread.
+ *
+ * <p>The returned callable is {@link ContinuationNativeCallable}, so it
runs inline inside the
+ * caller's continuation rather than on an async-pool worker, and its wait
suspends by yielding
+ * that continuation. A worker-dispatched wait would hold a pool worker
for the whole child call
+ * while the child's own durable calls queue behind it for a worker from
the same finite pool —
+ * a self-deadlock at {@code num-async-threads=1} and under enough nested
or concurrent waits at
+ * any size (issue #1213).
*/
@Override
protected DurableCallable<SubagentResult> prepare(
@@ -171,7 +181,7 @@ public class InternalSubagentSetup extends
BaseDeferredSubagentSetup {
throw new IllegalStateException(
"Failed to bootstrap internal sub-agent call for scope " +
scope, e);
}
- return new DurableCallable<SubagentResult>() {
+ return new ContinuationNativeCallable<SubagentResult>() {
@Override
public String getId() {
return sessionId + "#" + callId;
@@ -184,11 +194,11 @@ public class InternalSubagentSetup extends
BaseDeferredSubagentSetup {
@Override
public SubagentResult call() {
- // Runs off the mailbox thread, so the operator can dispatch
the child's
- // actions; failures converge into a failed SubagentResult
like the other
- // deferred setups.
+ // Runs inline on the mailbox thread; the wait yields the
continuation, so the
+ // operator can dispatch the child's actions without a pool
worker being held.
+ // Failures converge into a failed SubagentResult like the
other deferred setups.
try {
- return SubagentResult.ok(awaitSubagentCall(sessionId,
callId));
+ return SubagentResult.ok(awaitSubagentCallYielding(ctx,
sessionId, callId));
} catch (Exception e) {
return SubagentResult.error(e);
}
@@ -251,6 +261,29 @@ public class InternalSubagentSetup extends
BaseDeferredSubagentSetup {
* and returns its accumulated output. Must be invoked off the mailbox
thread.
*/
public List<Object> awaitSubagentCall(String sessionId, String callId)
throws Exception {
+ try {
+ return requireCallStatus(sessionId,
callId).getResponseFuture().get();
+ } catch (java.util.concurrent.ExecutionException e) {
+ throw unwrapChildFailure(e);
+ }
+ }
+
+ /**
+ * Awaits the identified call by yielding the caller's continuation, so
the wait holds no
+ * async-pool worker and the child can draw from the pool to make progress
(issue #1213).
+ *
+ * <p>The wait runs inline on the mailbox thread, so the blocking {@link
#awaitSubagentCall} is
+ * not an option here — it would pin the mailbox and stall the very child
it waits for. Only a
+ * Java action reaches {@link #prepare}; Python bootstraps through {@code
+ * RunnerContextImpl.bootstrapSubagentCallForScope} and blocks on its own
async worker instead.
+ */
+ private List<Object> awaitSubagentCallYielding(
+ RunnerContext ctx, String sessionId, String callId) throws
Exception {
+ InternalSubagentCallStatus cs = requireCallStatus(sessionId, callId);
+ return ((JavaRunnerContextImpl)
ctx).awaitCompletableFuture(cs.getResponseFuture());
+ }
+
+ private InternalSubagentCallStatus requireCallStatus(String sessionId,
String callId) {
InternalSubagentCallStatus cs = getCallStatus(sessionId, callId);
if (cs == null) {
throw new IllegalStateException(
@@ -259,17 +292,16 @@ public class InternalSubagentSetup extends
BaseDeferredSubagentSetup {
+ ", callId="
+ callId);
}
- try {
- return cs.getResponseFuture().get();
- } catch (java.util.concurrent.ExecutionException e) {
- // Surface the child's failure directly, so the caller's
SubagentResult carries
- // the failing action's exception rather than the future's
plumbing.
- Throwable cause = e.getCause();
- if (cause instanceof Exception) {
- throw (Exception) cause;
- }
- throw e;
- }
+ return cs;
+ }
+
+ /**
+ * Surfaces the child's failure directly, so the caller's {@link
SubagentResult} carries the
+ * failing action's exception rather than the future's plumbing.
+ */
+ private static Exception
unwrapChildFailure(java.util.concurrent.ExecutionException e) {
+ Throwable cause = e.getCause();
+ return cause instanceof Exception ? (Exception) cause : e;
}
/** The quiesce status of the identified call, or {@code null} when this
setup owns none. */
diff --git
a/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
b/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
index eb0b6003c..ef7e4f662 100644
---
a/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
+++
b/runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java
@@ -370,6 +370,44 @@ public class ContinuationActionExecutor {
: System.nanoTime() + timeout.toNanos();
}
+ /**
+ * Yields the current continuation until {@code future} completes, without
occupying an
+ * async-pool worker. The caller must be running inside a continuation on
the mailbox thread;
+ * this method yields that continuation and lets {@link #executeAction}
resume it once the
+ * future is done.
+ *
+ * <p>This is the non-blocking alternative to submitting a blocking {@code
future.get()} to the
+ * pool via {@link #executeAsync}. It prevents the deadlock where a parent
awaiting a child
+ * holds the only pool worker the child needs to make progress (issue
#1213).
+ */
+ public <T> T awaitFuture(ContinuationContext context, CompletableFuture<T>
future)
+ throws Exception {
+ context.clearAsyncState();
+ // Bridge to Future<?> so executeAction's hasPendingAsync gate drives
the resumption.
+ context.setPendingFuture(future.thenApply(v -> null));
+ // Re-checked before every yield, and resumption is polled by the
operator's mail loop
+ // rather than signalled by a completion callback, so a future already
done here — or one
+ // completing between the two lines above — simply never suspends.
+ while (!future.isDone()) {
+ Continuation.yield(SCOPE);
+ }
+ return joinCompletableFuture(future);
+ }
+
+ private static <T> T joinCompletableFuture(CompletableFuture<T> future)
throws Exception {
+ try {
+ return future.join();
+ } catch (CompletionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ } else if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw e;
+ }
+ }
+
/** Stops async work without waiting for worker exit cleanup. */
public void shutdown() {
asyncExecutor.shutdownNow();
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
index ed996a297..67ce7c7ec 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java
@@ -125,6 +125,29 @@ class JavaRunnerContextImplDurableExecuteAsyncTest {
assertTrue(persisted.isSuccess());
}
+ @Test
+ void testDurableExecuteAsyncContinuationNativeReconcilableRunsInline()
throws Exception {
+ InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
+ JavaRunnerContextImpl context = createContext(new ActionState(null),
executor);
+ TestContinuationNativeReconcilableCallable<String> callable =
+ new TestContinuationNativeReconcilableCallable<>(
+ "recon-inline",
+ String.class,
+ () -> "ok",
+ () -> fail("reconcile should not be called on initial
async execution"));
+
+ String result = context.durableExecuteAsync(callable).await();
+
+ assertEquals("ok", result);
+ assertEquals(1, callable.getCallCount());
+ assertEquals(0, callable.getReconcileCount());
+ assertEquals(0, executor.getExecuteAsyncCallCount());
+ assertEquals(2, persistCallCount.get());
+ CallResult persisted =
+
context.getDurableExecutionContext().getActionState().getCallResults().get(0);
+ assertTrue(persisted.isSuccess());
+ }
+
@Test
void testDurableExecuteAsyncReconcilableReplaySuccess() throws Exception {
InspectingContinuationActionExecutor executor = new
InspectingContinuationActionExecutor();
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/TestDurableCallables.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/TestDurableCallables.java
index 46ad2cedb..c1311cfd5 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/TestDurableCallables.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/TestDurableCallables.java
@@ -19,6 +19,7 @@
package org.apache.flink.agents.runtime.context;
import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.runtime.async.ContinuationNativeCallable;
import java.util.concurrent.Callable;
@@ -108,3 +109,20 @@ class TestReconcilableCallable<T> implements
DurableCallable<T> {
return reconcileCount;
}
}
+
+/**
+ * Reconciliable fixture that is also marked {@link
ContinuationNativeCallable}, so the runner
+ * context must run its {@link #call()} inline on the caller's thread rather
than dispatch it to an
+ * async-pool worker.
+ */
+class TestContinuationNativeReconcilableCallable<T> extends
TestReconcilableCallable<T>
+ implements ContinuationNativeCallable<T> {
+
+ TestContinuationNativeReconcilableCallable(
+ String id,
+ Class<T> resultClass,
+ Callable<T> callSupplier,
+ Callable<T> reconcileSupplier) {
+ super(id, resultClass, callSupplier, reconcileSupplier);
+ }
+}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentDeadlockTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentDeadlockTest.java
new file mode 100644
index 000000000..f2b0acd4f
--- /dev/null
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentDeadlockTest.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.subagent;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.api.OutputEvent;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.agents.AgentExecutionOptions;
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.SubagentFutures;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.api.subagent.SubagentSetup;
+import org.apache.flink.agents.plan.AgentConfiguration;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperator;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.condition.EnabledForJreRange;
+import org.junit.jupiter.api.condition.JRE;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reproduces the deadlock described in GitHub issue #1213: with {@code
num-async-threads=1}, a
+ * parent awaiting an internal sub-agent holds the single async-pool worker
via a blocking {@code
+ * responseFuture.get()}, while the child's own durable async call needs a
worker from the same pool
+ * to make progress. The result is a self-deadlock — neither side can advance.
+ *
+ * <p>The fix must let the parent's await yield its worker so the child can
proceed, regardless of
+ * pool size. This test times out under the broken behavior and completes
promptly once fixed.
+ */
+@EnabledForJreRange(min = JRE.JAVA_21)
+public class InternalSubagentDeadlockTest {
+
+ private static final AtomicBoolean CHILD_DURABLE_EXECUTED = new
AtomicBoolean(false);
+ private static final AtomicInteger DURABLE_EXECUTIONS = new
AtomicInteger();
+
+ @BeforeEach
+ void resetCounters() {
+ CHILD_DURABLE_EXECUTED.set(false);
+ DURABLE_EXECUTIONS.set(0);
+ }
+
+ /**
+ * Child agent whose action performs a durable async call before emitting
output. The call
+ * itself is trivial (returns immediately), but it still requires an
async-pool worker to
+ * execute. Under the deadlock, the parent holds the only worker and this
call never starts.
+ */
+ public static class DurableAsyncChildAgent extends Agent {
+ public DurableAsyncChildAgent() throws Exception {
+ addAction(
+ new String[] {InputEvent.EVENT_TYPE},
+ DurableAsyncChildAgent.class.getMethod(
+ "handle", Event.class, RunnerContext.class));
+ }
+
+ @SuppressWarnings("unused")
+ public static void handle(Event event, RunnerContext ctx) throws
Exception {
+ String result = ctx.durableExecuteAsync(durableCall()).await();
+ ctx.sendEvent(new OutputEvent(result));
+ }
+ }
+
+ /**
+ * Child that itself awaits a grandchild performing a durable call, so two
waits are nested
+ * while the grandchild's call still needs a worker from the same
single-thread pool.
+ */
+ public static class NestingDurableChildAgent extends Agent {
+ public NestingDurableChildAgent() throws Exception {
+ addResource("grandchild", ResourceType.AGENT, new
DurableAsyncChildAgent());
+ addAction(
+ new String[] {InputEvent.EVENT_TYPE},
+ NestingDurableChildAgent.class.getMethod(
+ "relay", Event.class, RunnerContext.class));
+ }
+
+ @SuppressWarnings("unused")
+ public static void relay(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup grandchild =
+ (SubagentSetup) ctx.getResource("grandchild",
ResourceType.AGENT);
+ SubagentResult result =
+ grandchild.submit(ctx,
InputEvent.fromEvent(event).getInput()).await();
+ ctx.sendEvent(new OutputEvent("relayed:" + firstOutput(result)));
+ }
+ }
+
+ private static DurableCallable<String> durableCall() {
+ return new DurableCallable<String>() {
+ @Override
+ public String getId() {
+ return "child-durable";
+ }
+
+ @Override
+ public Class<String> getResultClass() {
+ return String.class;
+ }
+
+ @Override
+ public String call() {
+ CHILD_DURABLE_EXECUTED.set(true);
+ DURABLE_EXECUTIONS.incrementAndGet();
+ return "durable-result";
+ }
+ };
+ }
+
+ private static String firstOutput(SubagentResult result) {
+ return String.valueOf(((List<?>) result.getResult()).get(0));
+ }
+
+ @SuppressWarnings("unused")
+ public static void callChild(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup child = (SubagentSetup) ctx.getResource("child",
ResourceType.AGENT);
+ SubagentResult result = child.submit(ctx,
InputEvent.fromEvent(event).getInput()).await();
+ ctx.sendEvent(new OutputEvent("parent:" + firstOutput(result)));
+ }
+
+ @SuppressWarnings("unused")
+ public static void callChildTwiceCombined(Event event, RunnerContext ctx)
throws Exception {
+ SubagentSetup child = (SubagentSetup) ctx.getResource("child",
ResourceType.AGENT);
+ SubagentFutures combined = child.submit(ctx,
"a").combine(child.submit(ctx, "b"));
+ List<SubagentResult> results = combined.awaitAll();
+ ctx.sendEvent(
+ new OutputEvent(
+ "parent:"
+ + firstOutput(results.get(0))
+ + ","
+ + firstOutput(results.get(1))));
+ }
+
+ /**
+ * With {@code num-async-threads=1}, the parent's await must not hold the
only worker. If it
+ * does, the child's durable call can never execute and this test times
out.
+ */
+ @Test
+ @Timeout(30)
+ void singleAsyncThreadDoesNotDeadlockWhenChildUsesDurableCall() throws
Exception {
+ List<Object> output = run(plan("callChild", new
DurableAsyncChildAgent()));
+
+ assertThat(CHILD_DURABLE_EXECUTED).isTrue();
+ assertThat(output).containsExactly("parent:durable-result");
+ }
+
+ /**
+ * Nesting multiplies the waits in flight: a child awaiting its own
grandchild must not consume
+ * a worker either, or one nested call deadlocks the single-thread pool.
+ */
+ @Test
+ @Timeout(30)
+ void singleAsyncThreadDoesNotDeadlockOnNestedCalls() throws Exception {
+ List<Object> output = run(plan("callChild", new
NestingDurableChildAgent()));
+
+ assertThat(CHILD_DURABLE_EXECUTED).isTrue();
+ assertThat(output).containsExactly("parent:relayed:durable-result");
+ }
+
+ /**
+ * Concurrent calls combined into one wait: {@code awaitAll} prepares both
calls (dispatching
+ * both children) before awaiting either, so both waits are in flight at
once and neither may
+ * hold a worker.
+ */
+ @Test
+ @Timeout(30)
+ void singleAsyncThreadDoesNotDeadlockOnCombinedCalls() throws Exception {
+ List<Object> output = run(plan("callChildTwiceCombined", new
DurableAsyncChildAgent()));
+
+ assertThat(DURABLE_EXECUTIONS.get()).isEqualTo(2);
+
assertThat(output).containsExactly("parent:durable-result,durable-result");
+ }
+
+ // --- helpers ---
+
+ /** Builds a plan whose single caller action is {@code callerAction}, with
one async worker. */
+ private static AgentPlan plan(String callerAction, Agent child) throws
Exception {
+ AgentConfiguration config = new AgentConfiguration();
+ config.set(AgentExecutionOptions.NUM_ASYNC_THREADS, 1);
+
+ Agent parent = new Agent();
+ parent.addResource("child", ResourceType.AGENT, child);
+ parent.addAction(
+ new String[] {InputEvent.EVENT_TYPE},
+ InternalSubagentDeadlockTest.class.getMethod(
+ callerAction, Event.class, RunnerContext.class));
+ return new AgentPlan(parent, config);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<Object> run(AgentPlan plan) throws Exception {
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
harness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(plan, true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ harness.open();
+ harness.processElement(new StreamRecord<>(1L));
+ ((ActionExecutionOperator<Long, Object>) harness.getOperator())
+ .waitInFlightEventsFinished();
+ return ((List<StreamRecord<Object>>) harness.getRecordOutput())
+
.stream().map(StreamRecord::getValue).collect(Collectors.toList());
+ }
+ }
+}
