sangkyoonnam opened a new pull request, #1239: URL: https://github.com/apache/flink-agents/pull/1239
Linked issue: #1217 ### Purpose of change #### User-visible outcome An ordinary cancellation in a mixed gather previously propagated before new durable outcomes were finalized. Durable children before the first cancellation in input order now retain their persisted success or failure when finalization succeeds, so recovery replays the prefix instead of repeating its side effects. #### Runtime flow Java `JavaRunnerContextImpl.resolveAsyncBatch` reserves durable slots and executes the batch as before. It finds the first ordinary cancellation, passes the preceding durable execution prefix and start flags to `finalizeExecutedOutcomes`, then rethrows without advancing the call index. An earlier durable cancellation can propagate first. Without an ordinary cancellation, full-batch finalization is unchanged. Python `_BatchAsyncExecutionResult.__await__` selects the equivalent prefix for an ordinary Java interruption. It calls `_finalize_batch_execution(..., advance_call_index=False)` before raising. The helper accepts a supplier prefix and advances the index only when requested. #### Key decisions - Reuse the existing finalizers to retain durable-only ordering and start-flag handling; a separate persistence loop would duplicate those rules. - Keep the batch base on cancellation. Saving an outcome does not complete its handle; re-awaiting rebuilds the plan from state. ### Behavioral Semantics #### Interaction decisions The table assumes recognized cancellations in child outcomes and successful finalization. Java recognizes interruption and `CancellationException` cause chains; Python recognizes Java interruption through `is_java_interruption`. Native Python control-flow cancellation can abort outcome collection before this prefix path. | First cancellation in input order | Durable slots before it | Durable slot at/after it | Call index | |---|---|---|---| | Ordinary child | Started executable slots finalized; matching terminal slots unchanged | New/PENDING slots stay PENDING; matching terminal slots unchanged | Stays at base | | Durable child | Started executable slots finalized; matching terminal slots unchanged | Cancelled and later new/PENDING slots stay PENDING; matching terminal slots unchanged | Stays at base | | None | Started executable slots finalized; matching terminal slots unchanged | (n/a) | Advances by the batch's durable count | Cancellation leaves later executed outcomes unfinalized. Ordinary-only batches and timeouts without cancellation are unchanged. #### Behavioral contracts 1. A started durable child before the first cancellation in input order retains its SUCCEEDED or FAILED outcome when serialization and persistence succeed. 2. Interrupted and later new/PENDING durable slots remain PENDING; matching terminal slots remain unchanged. 3. Cancellation preserves the batch's call-index base. A completed batch, including a timeout without cancellation, advances by its durable count. 4. Re-awaiting the same cancelled gather at that base replays the saved prefix and reruns unresolved ordinary children. Pending durable children follow the existing execution/reconciliation path. 5. Unless prefix finalization propagates an earlier cancellation, the ordinary exception propagates unchanged. Java restores the interrupt flag for interruption signals; a bare `CancellationException` does not set it. #### Failure behavior - A durable child cancellation in the prefix propagates first; its slot and later new/PENDING slots remain PENDING. - Non-cancellation serialization/finalization errors become in-memory failure outcomes, which are discarded when the ordinary cancellation propagates. The helper does not persist the replacement failure. Serialization failure leaves the slot PENDING; a persistence error can occur after state replacement, with no rollback. - Java rethrows cancellation from serialization/finalization. Python's existing catch converts those exceptions, including Java interruption, into in-memory failures and continues. This difference is unchanged. - Failure before outcome collection bypasses prefix finalization. ### Tests J = Java `testMixedGatherOrdinaryInterruptionKeepsEarlierDurableOutcomes`. P = Python `test_ordinary_java_interruption_propagates_without_caching`, both `mixed_batch` variants. | Contract | Tests and limits | |---|---| | 1 | J and P cover SUCCEEDED; no FAILED-prefix mixed-cancellation test | | 2 | J and P cover a later new slot; terminal suffix slots untested | | 3 | J and P assert the base after cancellation and advancement after retry | | 4 | J and P re-await the same handle and assert the saved child ran once and the ordinary child ran twice | | 5 | J and P assert exception identity; J additionally asserts interrupt restoration | The new Java test and both Python mixed-batch variants fail against `9af407ed`. Not verified: - JDK 21 continuations with real task cancellation and checkpoint restore; these regressions re-await handles using test executors. - A durable cancellation before an ordinary one in the same mixed batch. - Cancellation in a mixed batch with a failed or empty prefix, unstarted durable slots, terminal slots, a nonzero base, or a timeout. - An ordinary Java `CancellationException` in a mixed batch. - Serialization/persistence failure during mixed-cancellation prefix finalization. <details> <summary>Author-reported verification and negative control</summary> - JDK 17: `JavaRunnerContextImpl*Test` 52 passed. - JDK 21: `JavaRunnerContextImpl*Test` and `ActionExecutionOperatorTest` 124 passed, 14 skipped (JDK < 21-only cases). - Python 3.12: `test_durable_exception.py` and `test_flink_runner_context_reconcilable.py` 118 passed. - `tools/lint.sh -f` / `-c` on JDK 11: clean. - With the two main-source files reverted to `9af407ed` and the new tests kept: the Java test fails on the first durable slot (`isSuccess` false), and both Python `mixed_batch` cases fail. </details> ### API No public API or persisted-format change. Successfully finalized durable outcomes before an ordinary cancellation are now replayed on recovery. Older PENDING slots still use the existing execution/reconciliation path. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Claude Code 2.1.296 (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]
