sangkyoonnam opened a new pull request, #1318: URL: https://github.com/apache/flink-agents/pull/1318
Linked issue: #1311 ### Purpose of change #### User-visible outcome On JDK 21+ continuation execution, `max-in-flight-input-records` now bounds the input records each Agent operator subtask has in flight. Queued Java async calls now favor records that started processing earlier, so a record's follow-up call no longer waits behind every later record's first call. With the Java ReAct Quickstart on JDK 21 and the default limit of 100, the first result now prints after 62 s; before, nothing printed in 10 minutes. #### Runtime flow Admission: on continuation execution, `processElement` yields to the mailbox while `inFlightInputRecords` is at the limit, then sets the incoming record's key again. An input that triggers no Actions now keeps its key busy until it retires. Ordering: record indices follow processing start order, including after a busy-key wait or recovery, and last until retirement. A Java Action keeps its first-dispatch `(recordIndex, taskIndex)` in its `ContinuationContext` across resumes. The JDK 21 `ContinuationActionExecutor` queues `executeAsync` and `executeAllAsync` work in a `PriorityBlockingQueue` by that pair, then by submission. #### Key decisions - Reuse the existing option and the parallel engine's admission wait. - Reuse the parallel engine's `(recordIndex, taskIndex)` order in the async executor's work queue, where external calls wait; Actions keep that priority across resumes. Mailbox mails stay FIFO, without its preference for resuming existing Actions, so a mailbox backlog can still delay a follow-up call's submission. I can add it here if you'd like. - Serial execution, including Python Actions on JDK 11–20, still ignores the limit. Changing its admission requires separate JDK 17 and Python coverage. ### Behavioral Semantics #### Interaction decisions | Engine selected | Limit | Async queue order | |---|---|---| | Parallel engine (JDK < 21) | Enforced, unchanged | Unchanged | | Continuation execution (JDK 21+) | Ignored → enforced | Submission → record start, Action dispatch, submission | | Serial execution | Ignored, unchanged | Unchanged | #### Behavioral contracts 1. On continuation execution, a new input waits while `max-in-flight-input-records` admitted inputs, including pending and recovered ones, are unfinished. 2. A record admitted after waiting is processed under its own key. 3. An input that triggers no Actions keeps its key busy, and its budget, until it retires. 4. On continuation execution, queued Java async work of a record that started earlier runs first; within a record, an Action dispatched earlier goes first, then submission order. 5. Parallel execution is unchanged. Serial execution still ignores the limit but gets the busy-key fix. #### Failure behavior As on the parallel engine, the admission wait has no timeout, so a record that never finishes blocks admission. Running calls aren't preempted, and work without an operator-assigned priority sorts last. ### Tests | Contract | Coverage | |---|---| | 1 | `ParallelExecutionE2ETest.maxInFlightInputRecordsThrottlesAdmission`, which no longer skips JDK 21+; `nonParallelEnginesRetireInputsWithoutTriggeredActions` | | 2 | `continuationExecutionProcessesRecordAdmittedAtInFlightLimitUnderItsOwnKey` | | 3 | `nonParallelEnginesRetireInputsWithoutTriggeredActions`, `nonParallelEnginesKeepKeyBusyUntilNoActionInputRetires`, `nonParallelEnginesAdvancePastPendingInputWithoutTriggeredActions` | | 4 | `ContinuationActionExecutorPriorityTest.queuedWorkRunsInRecordThenTaskOrder`; `nonParallelEnginesAssignPrioritiesByRecordStartThenFirstDispatch` | | 5 | Existing parallel engine tests; test 1 above on JDK 17 | Unqualified tests are in `ActionExecutionOperatorTest`. Unverified on continuation execution: restore with pending inputs, and operator-level ordering across a resumed Action. <details> <summary>Test runs and Quickstart measurements</summary> - The CI matrix, zero failures and errors: `tools/ut.sh -j` on JDK 11, 17 and 21 (runtime 1,467, 1,467 and 1,475 tests; 51, 51 and 70 skipped), and `tools/ut.sh -j -e` with Flink 2.0, 2.1 and 2.2 on JDK 21 (71 tests, 16 skipped, each) and with 1.20 and 2.3 on JDK 17 (67, 20 skipped, each). - `queuedWorkRunsInRecordThenTaskOrder` submits in reverse order; with the previous FIFO executor, `record 2` runs first. - Java ReAct Quickstart, Flink 2.3.0, JDK 21, Ollama `qwen3:8b` with two slots, default limit, Trace Log times: first result after 62 s, 15 results in 10 minutes. On main at `60bf8cd0`, 3,033 runs started and nothing printed in 10 minutes. The Python ReAct Quickstart in the same setup printed its first result after 186 s and 19 within 10 minutes, with no runtime errors; its 3,163 reviews share 31 keys. </details> ### API No public signature changes. By default, continuation execution now admits at most 100 unfinished inputs per subtask instead of all of them. ### Documentation - [ ] `doc-needed` - [ ] `doc-not-needed` - [x] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Claude Code 2.1.303 (Claude Opus 5.5) Generated-by: Claude Code 2.1.307 (Claude Opus 5.5) -- 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]
