wenjin272 opened a new pull request, #1361: URL: https://github.com/apache/flink-agents/pull/1361
Linked issue: Closes #1360 ### Purpose of change Java serial execution and Python execution on JDKs below 21 now honor `max-in-flight-input-records`. Previously, these paths could admit records beyond the configured limit. #### Runtime flow Before admitting an input, `ActionExecutionOperator.processElement` checks the in-flight count. At capacity, it yields to the mailbox until a record finishes, then restores the incoming record's key and admits it. The existing counter includes both active inputs and inputs queued for a busy key. #### Key decisions Reuse the existing mailbox admission loop for every non-worker path by removing its continuation-only guard. This keeps completion processing available while waiting and avoids introducing another queue or counter. The parallel worker path retains its existing lock-aware admission loop. ### Behavioral Semantics #### Interaction decisions | Execution path | Admission behavior | |---|---| | JDK < 21, Java, parallel disabled | Newly enforces the cap through mailbox yielding | | JDK < 21, Python | Newly enforces the cap through mailbox yielding | | JDK < 21, Java, parallel enabled | Existing lock-aware cap enforcement | | JDK 21 continuation path | Existing mailbox-based cap enforcement | #### Behavioral contracts - Each subtask counts every admitted, unfinished input toward the limit, including same-key queued inputs. The default remains 100. - At capacity, new input admission waits while existing work can complete and release capacity, including inputs that trigger no Actions. - Inputs admitted after yielding execute under their own key. - The limit permits overlap across independent keys when capacity is available, while preserving same-key sequencing. - Backpressure at capacity can delay checkpoint barriers; this change does not guarantee checkpoint progress under sustained saturation. #### Failure behavior Non-positive limits continue to fail during operator initialization. Exceptions from mailbox processing propagate through `processElement`; there is no new fallback, retry, or exception suppression. Existing Action failure handling is unchanged. If admitted work never completes, admission can remain blocked. ### Tests | Contract | Coverage | |---|---| | Active and queued inputs consume capacity; default 100 | `ActionExecutionOperatorTest.inputLimitCountsBothActiveAndQueuedRecords` | | Waiting permits completion, including inputs without Actions | `nonParallelEnginesRetireInputsWithoutTriggeredActions`; `ParallelExecutionE2ETest.maxInFlightInputRecordsThrottlesAdmission` | | Correct input key after yielding | `nonParallelEnginesProcessRecordAdmittedAtInFlightLimitUnderItsOwnKey`; Python per-key counters | | Independent-key overlap and same-key sequencing | Python `test_input_limit_controls_async_overlap`, caps 1 and 2 | | Invalid limits fail initialization | Existing `openRejectsNonPositiveMaxInFlightInputRecords` | | Barrier delays under sustained saturation | Not directly tested in this change | Local validation passed: - JDK 17 full runtime suite: 1,508 passed, 48 skipped. - Final operator test class: JDK 11, 86 passed; JDK 21, 72 passed and 14 expected skips. - Java admission E2E: both serial/parallel cases passed on JDK 11 with Flink 1.20.5 and JDK 17/21 with Flink 2.3.0. - Python admission E2E: both caps passed on JDK 17/21 with Flink 2.3.0, using freshly built local distribution JARs. - Java/Python formatting and license checks passed; the subsequent edits only simplified comments and documentation. Not verified: Python on JDK 11, checkpoint/recovery under a saturated input limit, and the full repository test suite. The final default-limit test case was added after the full JDK 17 runtime run and passed in the final JDK 11/21 operator runs. <details> <summary>Regression evidence and implementation invariants</summary> With only the production operator reverted to its previous implementation, five of seven selected regression cases failed; the fixed implementation passed. This covers serial explicit/default caps, inputs with no triggered Actions, and keyed admission after yielding. The incoming key is set after the final mailbox yield because completion mail may change the current key. Existing increment, retirement, and restore accounting are reused. JDK 21 validation used a freshly packaged multi-release runtime JAR containing the Java 21 continuation executor. </details> ### API No API signatures, configuration keys, defaults, or state formats change. Existing low-JDK Java serial and Python jobs now apply the configured limit, including the default of 100, and may experience backpressure where input admission was previously unbounded by this option. ### 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]
