eemario opened a new pull request, #1147:
URL: https://github.com/apache/flink-agents/pull/1147
Execute plans whose actions are implemented in Java on coordinator workers
when continuation-based execution is unavailable. Shared operator state remains
protected by the parallel execution lock, while blocking asynchronous calls
release the lock and restore task context after worker re-acquisition.
Runtime flow:
The mailbox thread admits each input record under the in-flight budget and
appends its action tasks to the per-key queue. The coordinator grants workers
the parallel execution lock in ascending (recordIndex, taskIndex) order, with
absolute priority for the mailbox thread. A worker pulls its task from keyed
state, prepares and runs the action; a blocking asynchronous call releases the
lock before awaiting and restores the task context after re-acquisition.
Commits happen on the mailbox thread, so keyed state is only ever mutated under
the lock.
Key decisions:
- Worker-pool engine on JDK < 21, where continuation-based execution is
unavailable; engine eligibility is validated at setup and ineligible plans fall
back to the previous engine.
- Drain in-flight action tasks at prepareSnapshotPreBarrier instead of
snapshotting them: pulled-but-uncommitted results live only in memory and would
be lost on restore, so the barrier waits until the engine is quiesced and the
snapshot captures no half-pulled intermediate state. Dispatch resumes only
after the synchronous snapshot capture, because a worker woken any earlier
could pull a queued task out of keyed state before it is captured. Restore
semantics are therefore identical to the serial engine.
- Mailbox-priority lock, so checkpoint and commit work on the mailbox thread
is never starved by workers.
- Bound concurrent input records with in-flight backpressure instead of
unbounded admission.
- Align lifecycle notifications with the serial path:
prepared/started/reused fire in execute(), finishing/finished/ failed fire in
commit(), with noop-silent guards.
- Register the durable-execution context in the lookup map on setup so
workers resolve it during replay.
Interaction decisions (engine selection):
Plan parallel-execution.enabled Engine
pure-Java, JDK < 21 true (default) parallel
pure-Java, JDK < 21 false serial
coroutine engine available any (ignored) coroutine
contains Python actions any (ignored) serial
Behavioral contracts:
- Same-key input records commit in input order.
- Fan-out siblings, broadcast listeners and multi-key inputs run
concurrently; each action observes its own action-scoped context.
- At most max-in-flight-input-records (default 100) records are in flight;
admission then blocks until an in-flight record retires.
- A checkpoint contains no in-flight execution state; after restore the
engine behaves exactly as the serial path and exactly-once output is preserved.
- A failing action task fails the job on the mailbox thread without hanging.
- Task lifecycle notifications are indistinguishable from the serial path.
Failure behavior:
- Pull/prepare/run failures on a worker are captured and rethrown on the
mailbox thread; the lock is always returned and no pool thread dies silently.
- A failed prepareSnapshotPreBarrier rolls the drain back via stopDraining
and resumes dispatch, because such a failure never reaches snapshotState's
stopDraining.
- Aborting an old checkpoint does not resume a drain belonging to the
current checkpoint.
- Unsupported configurations (Python actions, coroutine engine) are not
errors: the parallel options are ignored and execution falls back to the
previous engine.
Compatibility impact:
- Adds AgentExecutionOptions.PARALLEL_EXECUTION_ENABLED (default true) and
MAX_IN_FLIGHT_INPUT_RECORDS (default 100) to the Java and Python APIs; no
existing option changes meaning.
- Pure-Java agents on JDK < 21 switch from serial to parallel execution by
default; per-key ordering and exactly-once semantics are unchanged, and
parallel-execution.enabled=false restores the previous behavior.
- JDK 21 coroutine execution and plans containing Python actions are
unaffected.
Contracts to tests:
- Same-key order: ParallelExecutionE2ETest.sameKeyInputsCommitInOrder;
ParallelExecutionTaskQueueTest
- Concurrent fan-out/broadcast/multi-key and action-scoped context:
ParallelExecutionE2ETest.fanoutRunsSiblingActionsInParallel,
broadcastRunsAllListenersInParallel, multiKeyInputsRunInParallel,
parallelSiblingsKeepActionScopedContext; ActionTaskContextManagerTest
- In-flight bound: ParallelExecutionE2ETest.
maxInFlightInputRecordsThrottlesAdmission; ParallelExecutionCoordinatorTest
- Checkpoint consistency and recovery:
ActionExecutionOperatorDrainDeadlockIntegrationTest;
ActionExecutionOperatorCapacityIntegrationTest.
actionsBeyondWorkerCountRecoverFromCheckpointWithExactlyOnceOutput;
ParallelExecutionLockTest
- Failure propagation: ParallelExecutionE2ETest. asyncFailureFailsTheJob,
failingSiblingFailsTheJobWithoutHanging
- Lifecycle notifications: TaskLifecycleListenerNotificationTest;
ActionExecutionOperatorTest
- Lock handoff on blocking async calls: ContinuationActionExecutorTest;
JavaRunnerContextImplDurableExecuteAsyncTest
- Beyond-worker capacity: ParallelExecutionE2ETest.
tasksBeyondWorkerCountCompleteExactlyOnce;
ActionExecutionOperatorCapacityIntegrationTest
Not verified: plans containing Python actions (the engine never applies by
design); JobManager failover while a drain is in progress.
Generated-by: Qoder
<!--
* Thank you very much for contributing to Flink Agents.
* Please add the relevant components in the PR title. E.g., [api],
[runtime], [java], [python], [hotfix], etc.
-->
<!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this.
-->
Linked issue: #1119
### Purpose of change
<!-- What is the purpose of this change? -->
### Behavioral Semantics
<!-- For a non-trivial code change whose implementation is largely
AI-assisted: interaction decisions, behavioral contracts, and failure behavior.
See `contribution-guides/ai-assisted-pr.md`. Remove this heading and this
comment otherwise. -->
### Tests
<!-- How is this change verified? -->
### API
<!-- Does this change touches any public APIs? -->
### Documentation
<!-- Do not remove this section. Check the proper box only. -->
- [ ] `doc-needed` <!-- Your PR changes impact docs -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-included` <!-- Your PR already contains the necessary
documentation updates -->
### Was this patch authored or co-authored using generative AI tooling?
<!-- Do not remove this section. Check the proper box only. -->
- [ ] Yes
- [ ] No
If yes, include a `Generated-by: <tool name and version> (<model name and
version>)` line, for example `Generated-by: Claude Code 2.1.226 (Claude Opus
4.6)`, in the commit message so it reaches Git history. Repeat the same line
here for reviewer visibility. See the [ASF generative tooling
guidance](https://www.apache.org/legal/generative-tooling.html).
--
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]
