wenjin272 opened a new issue, #1122: URL: https://github.com/apache/flink-agents/issues/1122
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description This is a child issue of #1055. It tracks a review of the public asynchronous durable execution APIs before the 0.4 compatibility baseline. #### Background Java and Python currently expose substantially different models for asynchronous durable execution. Java exposes a synchronous-looking API even for asynchronous execution: ```java <T> T durableExecuteAsync(DurableCallable<T> callable) throws Exception; ``` The runtime submits the callable to a worker pool and uses a Continuation to yield the Action, but the caller receives the final value rather than a future. As a result, individual calls cannot be created first and then composed into a concurrent batch. Python returns an awaitable: ```python result = await ctx.durable_execute_async( func, *args, durable_id="logical-call-id" ) ``` The current `AsyncExecutionResult` is deferred: execution is submitted when it is awaited. It does not support standard `asyncio.gather()`. Parallel execution therefore uses the separate `durable_execute_all_async()` API introduced in #926, which requires users to construct `DurableCall` descriptors. The APIs differ not only syntactically, but also in what "async" means: - Java hides suspension internally and returns `T`. - Python exposes a deferred awaitable. - Batch execution uses a second invocation form rather than composing single-call handles. This also complicates the `durable_id` discussion in #1016: the single-call Python API already accepts `durable_id`, while the public `DurableCall` batch descriptor needs a separate way to carry or wrap the same identity. #### Proposed direction Make an asynchronous durable invocation a first-class, composable handle in both languages while keeping the surface idiomatic for each language. Illustrative Python API: ```python first = ctx.durable_execute_async(foo, x, durable_id="foo-x") second = ctx.durable_execute_async(bar, y, durable_id="bar-y") outcomes = await ctx.gather(first, second) ``` `DurableCall` may remain an internal batch-planning representation, but users should not need to construct it. Each single-call handle already carries its recovery identity and invocation arguments. Illustrative Java API: ```java DurableFuture<T> first = ctx.durableExecuteAsync(call1); DurableFuture<T> second = ctx.durableExecuteAsync(call2); List<Outcome<T>> outcomes = ctx.await(ctx.gather(List.of(first, second))); ``` For example, the corresponding contracts could be: ```java <T> DurableFuture<T> durableExecuteAsync(DurableCallable<T> callable); <T> T await(DurableFuture<T> future) throws Exception; <T> DurableFuture<List<Outcome<T>>> gather( List<? extends DurableFuture<T>> futures); ``` The exact names and types are open for discussion. A framework-managed handle may be preferable to exposing a raw `CompletableFuture`: calling `Future.get()` or `CompletableFuture.join()` would block the current thread and bypass the Continuation yield mechanism. `RunnerContext.await()` can instead provide the Java equivalent of Python's `await` and yield the current Action while the operation is incomplete. `gather` should be a composition operation: it combines multiple durable handles into one aggregate handle. It should not require a second representation of `durable_id` or other per-call metadata. #### Semantics to define The API review should explicitly decide and document: - Whether creating a durable async handle submits the operation immediately or defers submission until `await` / `gather`. - When a durable slot is assigned and persisted, especially when several handles are created before any of them are awaited. - How Java `await` integrates with JDK 21 Continuation execution without blocking the mailbox thread, and what the JDK 11 fallback does. - Whether an Action may finish with unresolved durable handles, or whether this should fail validation. - Result ordering and per-call failure behavior for `gather`. - Timeout, cancellation, and reconciler behavior for individual and gathered calls. - Recovery after partial batch completion: completed calls should be replayed independently and only incomplete calls should execute or reconcile again. - How the accepted identity semantics from #1016 apply uniformly to single and gathered calls. #### Compatibility and migration This is likely a breaking Java API change because `durableExecuteAsync()` currently returns `T`. Existing call sites would migrate from: ```java T result = ctx.durableExecuteAsync(callable); ``` to: ```java T result = ctx.await(ctx.durableExecuteAsync(callable)); ``` The existing `durableExecuteAllAsync()` APIs could be deprecated, retained as convenience wrappers, or replaced after the new composition model has equivalent recovery, timeout, ordering, and failure semantics. #### Acceptance criteria - Agree on the Java and Python public API shapes and their shared semantic contract. - Java asynchronous durable execution returns a composable handle and provides a runtime-aware, non-blocking wait operation. - Python can compose single durable invocations with `gather` without requiring users to construct `DurableCall`. - Single and gathered calls use the same per-call durable identity mechanism. - Tests cover sequential waiting, actual parallel execution, deterministic ordering, individual failures, timeout/cancellation, recovery with partial completion, and unresolved handles. - JDK 11 fallback and JDK 21 Continuation behavior are both covered. - Documentation and migration examples are updated for both languages. ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
