wenjin272 opened a new pull request, #1315: URL: https://github.com/apache/flink-agents/pull/1315
Linked issue: Fixes #1311 ### Purpose of change JDK 21 continuation execution now honors `max-in-flight-input-records` and favors queued asynchronous calls from earlier active input runs. A ReAct run's follow-up calls can therefore proceed ahead of later runs' queued first calls instead of being trapped behind the entire FIFO backlog. #### Runtime flow At the input limit, `ActionExecutionOperator` processes mailbox work until an input retires, then restores the incoming record's key before admission. Active and pending inputs both consume budget. Inputs with no matching Actions release their budget, and their key stays busy until retirement. Each active continuation input receives an activation ordinal. Each Java Action receives a dispatch ordinal when its continuation context is first prepared; the context retains it across suspension and resumption. Single calls and `gather` submissions enter the same priority queue, ordered by input, Action, then submission ordinal. Follow-up and internal child Actions share the active input's priority. #### Key decisions Apply the admission change to continuation execution; leave serial admission and the existing parallel engine unchanged. Prioritize the actual async work queue, where model calls accumulate. Rebuild transient scheduling priorities during recovery instead of changing checkpoint formats. Same-key pending inputs receive a new priority when their next run becomes active. ### Behavioral Semantics #### Interactions | Execution path | Input admission | Asynchronous scheduling | | --- | --- | --- | | JDK 21 continuation, Java Actions | Enforces the configured cap | Queued single/batch calls use input and Action priority | | JDK 21 continuation, plans with Python Actions | Enforces the same operator cap | Python-native scheduling is unchanged; Java calls using the continuation executor use its priority queue | | Pre-JDK 21 parallel engine | Existing cap remains | Existing scheduler remains | | Serial fallback | Continues to ignore the cap | Existing synchronous behavior remains | #### Contracts - Admission waits while admitted, unfinished inputs reach the cap; pending same-key inputs count too. - Waiting does not change the incoming record's key. A no-Action input releases its budget and cannot overlap the next input for its key. - Recovery reconstructs active and pending budget, including when restoring with a smaller cap. - Earlier active inputs' queued Java calls take precedence over later inputs' queued calls, including follow-up Actions and `gather`. Equal priorities retain submission order. - Running calls are not preempted. Output ordering and checkpoint-stable cross-key scheduling order are not guaranteed. #### Failure behavior Non-positive limits remain invalid. Exceptions from mailbox work and external calls continue through the existing failure paths; this change adds no retries or fallback. Existing batch timeout/cancellation handling is retained. Priorities are scheduling metadata and do not alter durable call identities or stored outcomes. ### Tests | Contract | Regression coverage | | --- | --- | | Admission waits for retirement | `backpressureBlocksAdmissionAtCapAndReleasesOnRetirement` | | Incoming key survives mailbox waiting | `continuationAdmissionRestoresIncomingKeyAfterWaiting` | | No-Action budget release and same-key serialization | `concurrentEnginesRetireInputWithoutTriggeredActions`, `continuationKeepsNoActionKeyBusyUntilItsInputRetires` | | Active/pending budget after restore | `continuationRestoreCountsActiveAndPendingInputsBeforeNewAdmission` | | Follow-up priority propagates through the operator | `continuationFollowUpActionInheritsInputPriority` | | Single/batch priority and stable queue ties | `ContinuationSchedulingTest` | Validation: - JDK 21 `./tools/ut.sh -j` before the final Action-priority refinement: 3,547 tests, 77 skipped, zero failures/errors. Final-source focused results are listed below. - Final-source JDK 21 focused runtime regression: 388 tests, 20 skipped, zero failures/errors; includes subagents, durable execution, and Python bridges. - Final-source JDK 11 focused regression: 175 tests, 7 skipped, zero failures/errors; includes the legacy parallel engine and serial fallback. - Five admission/priority regression cases fail against the original operator/executor and pass with the fix. - JDK 11 Spotless, Ruff for the edited Python file, and `git diff --check` pass. Not verified: the reporter's live Ollama Quickstart timing, deployed Flink-cluster E2E behavior, Python-native scheduling priority (unchanged), or fairness under indefinitely running input chains. <details> <summary>Build and verification details</summary> Focused runs use fresh source copies and `mvn -pl runtime -am clean test` with the relevant test selectors and `-Dsurefire.failIfNoSpecifiedTests=false`. `JAVA_HOME` selects each JVM; Python-facing runs set `PYTHONPATH` to the active environment's site-packages. The Java script rebuilds and installs reactor modules before testing. The JDK 21 test overlay exercises the real continuation implementation. Transient active-input priorities also keep a no-Action key busy until its retirement mail runs. Queue entries capture immutable priorities at submission. First-time Action preparation creates its contexts before checking whether it needs a new priority; generated continuation tasks reuse the transferred context. Restored keys receive priorities before their work is scheduled. </details> ### API No public API or configuration key/default changes. Existing JDK 21 users now receive the configured backpressure (default 100 inputs), and queued Java calls may execute in a different order. Checkpoint formats are unchanged. Serial and pre-JDK 21 parallel execution retain their admission policies. ### 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: 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]
