wenjin272 opened a new pull request, #1208: URL: https://github.com/apache/flink-agents/pull/1208
Linked issue: #1191 ### Purpose of change Add deferred `ctx.execute_async` (Python) / `ctx.executeAsync` (Java) for work whose results do not need recovery persistence. Ordinary and durable calls can be combined with `ctx.gather`. Java batches also execute concurrently on JDKs below 21. #### Runtime flow Creating a handle does not execute its callback. Await resolves it through the runtime; gather validates its children, reserves durable slots, schedules uncached calls, and returns input-ordered outcomes. Only durable results enter recovery state. On JDK < 21, a parallel Action worker releases the execution lock while running a single callback inline or waiting for gather's separate batch pool, then restores its context under the lock. Without the parallel Action engine, the caller blocks while batch workers still run concurrently. JDK 21 uses the continuation scheduler and its async executor. Python uses its existing shared executor and runtime-aware waiting. #### Key decisions Both deferred future types share one waiting interface. A separate old-JDK batch pool avoids starvation when all Action workers await children. Both pools receive shutdown signals before either is joined, preventing Python interpreter cleanup from waiting on a callback that has not been interrupted. ### Behavioral Semantics #### Interaction decisions | Calls / condition | Behavior | |---|---| | Ordinary, first await | Execute without serialization or durable slots | | Durable, recovery | Reuse persisted outcomes or follow existing pending-call reconciliation | | Mixed gather, recovery | Re-execute ordinary calls; reserve/replay only durable children | | Cached child in gather | Reuse local outcome without dispatch | | Java JDK < 21 / JDK 21+ | Separate batch pool / continuation async pool | | Python, any JDK | Existing Python worker pool | #### Behavioral contracts 1. Handles are cold; repeated awaits reuse their local result or business exception. 2. Ordinary results need not be serializable and are recomputed after Action replay. 3. Mixed batches preserve durable replay identity and existing reconciliation semantics. 4. Gather returns one outcome per input in input order, with concurrency bounded by `async.batch.parallelism` and available executor workers. 5. Gather rejects foreign, duplicate and nested batch handles before dispatch. 6. The batch deadline includes queue waiting. Completed outcomes survive timeout; unstarted durable calls remain pending for recovery. 7. Operator shutdown signals both Java pools before joining them and closes resources after worker cleanup. #### Failure behavior Single await raises callback errors; gather records ordinary callback errors per outcome. Interruption/cancellation propagates without caching a terminal business failure; Java fatal `Error`s propagate. Invalid batch inputs raise. Parallelism is normalized to at least one; a nonpositive timeout disables the deadline. Timeout fails unfinished outcomes and prevents queued/unsubmitted callbacks from starting. Already-running callbacks may continue and produce side effects; they need their own I/O timeouts. Started durable calls retain timeout failures. Shutdown interrupts workers and aggregates cleanup failures while attempting remaining cleanup; uninterruptible callbacks can still delay shutdown. Callbacks must not mutate context, memory or events; obtain resources first and use thread-safe clients. No public cancellation API is added. ### Tests Passed: **334 focused Java tests on JDK 11**, **113 Python tests**, **3 real JDK 21 continuation tests**, and **4 Flink batch E2E tests on JDK 17**. Spotless, Ruff and changed-file license checks pass. Coverage targets scheduling, replay, timeouts and shutdown. | Contract | Tests | |---|---| | 1. Deferred execution and local caching | Java ordinary single/batch tests; Python deferred and cached-failure tests | | 2. No ordinary persistence; recompute on replay | Java `testOrdinaryAsyncIsDeferredAndDoesNotSerializeOrPersist`; Java/Python mixed recovery tests | | 3. Durable identity and reconciliation | Java `testMixedBatchMapsCachedResultPendingReconcilerAndFailureToDurableSlots`; Python mixed recovery test | | 4. Ordering, concurrency and cap | `AsyncBatchExecutorTest`; Java runtime scheduler tests on JDK 11/21; Python ordinary/mixed concurrency tests; batch E2E | | 5. Reject invalid handles before execution | Java/Python invalid-handle tests | | 6. Completed/started/unstarted timeout handling | Java/Python timeout recovery tests; executor queued/unsubmitted test; batch E2E | | 7. Shutdown ordering and cleanup failures | Operator Python-batch shutdown regression, cleanup-failure test, executor close test | Not verified: full repository/external-service suites; live checkpoint/failover E2E for mixed calls; full Flink E2E on JDK 21. The shutdown regression mocks the interpreter, retaining the real lifecycle manager. Context restrictions and client thread safety are caller obligations. <details> <summary>Implementation invariants and validation details</summary> - Durable indices are reserved in relative durable-input order before dispatch. Ordinary children consume no indices. State finalization occurs after restoring the Action lock/context. - Old-JDK batches use a sliding window, completion queue and synchronized start gate. The shared pool is per operator subtask; `num-async-threads` sizes it independently of the Action worker pool. - Java: focused Maven reactor tests across plan/runtime (98 + 236). JDK 21 regression explicitly compiled and loaded `src/main/java21` ahead of base classes and asserted continuation support. - Python: core option, runtime context and tool-action pytest suites with the virtual environment's purelib on `PYTHONPATH`. - E2E: JDK 17 targeted `clean install -DskipTests` through the integration-test module, then `AsyncExecutionTest#testToolCallBatch*`, avoiding stale shaded artifacts. - Shutdown regression timed out before the fix and passed afterward. Changed-file RAT reports zero unknown licenses. </details> ### API Adds `AsyncFuture` and ordinary execution methods in both languages; `DurableFuture` extends `AsyncFuture`. Java gather now accepts `AsyncFuture` children and returns `AsyncFuture<List<Outcome<T>>>`; callers expecting a durable batch return type and custom RunnerContext implementations need updating. Renames `tool-call.parallelism` / `tool-call.batch.timeout.ms` to `async.batch.parallelism` / `async.batch.timeout.ms`, and constants to `ASYNC_BATCH_PARALLELISM` / `ASYNC_BATCH_TIMEOUT_MS`, without aliases. Defaults and `tool-call.async` are unchanged. Older JDKs add a bounded batch pool. Existing durable single-call persistence semantics remain unchanged. ### Documentation - [ ] `doc-needed` - [ ] `doc-not-needed` - [x] `doc-included` Workflow, tool-use and configuration docs cover both APIs and renamed options. ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Codex CLI 0.153.4 (GPT-6) -- 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]
