uvashishtha opened a new pull request, #29585:
URL: https://github.com/apache/flink/pull/29585
## What is the purpose of the change
Since FLINK-33768 (FLIP-379, 1.19), `ExecutionJobVertex` creates its
operator coordinators in its constructor, but
`DefaultOperatorCoordinatorHandler` only tracks those of initialized vertices
(`filter(ExecutionJobVertex::isInitialized)`), adding a vertex's coordinators
when it is initialized. `disposeAllOperatorCoordinators`, which
`SchedulerBase#closeAsync` calls when the JobMaster stops, therefore never
closes the coordinators of vertices that are not initialized yet.
With the `AdaptiveBatchScheduler`, a source vertex stays uninitialized while
its `SourceCoordinator` runs `DynamicParallelismInference#inferParallelism`.
When the job is cancelled or fails during inference:
- the inference is not interrupted and runs to completion after the job has
ended;
- the `SourceCoordinator`'s context is never closed, so its
`SourceCoordinator-<name>` executor thread leaks.
On a MiniCluster, the job reached CANCELED in 28 ms and the JobMaster
stopped 3 ms later, but a 20 s inference ran to its end and its coordinator
thread was still alive afterwards.
## Brief change log
- `disposeAllOperatorCoordinators` closes the coordinators of every vertex
in the execution graph instead of those in the handler's `coordinatorMap`. The
map's entries all come from the same graph's vertices, so this closes a
superset of what the map held, each holder once. The map keeps its other jobs
(event routing, start, lazy initialization).
- New `ExecutionJobVertex#getOperatorCoordinatorsEvenIfUninitialized`.
`getOperatorCoordinators` keeps its `isInitialized` check (pinned by
`ExecutionJobVertexTest#testLazyInitialization`).
This also covers a vertex that counts as initialized but whose coordinators
were never registered, which happens when `initialize()` fails after it
assigned the task vertices.
For a source, closing shuts down the coordinator executor and so interrupts
the inference (`SourceCoordinator#close` closes its context even when never
started, since FLINK-32411). The interrupt only helps an inference that
responds to it: an inference that ignores interrupts still outlives the job,
though it no longer holds up the JobMaster, since the holder closes on its own
thread with a timeout.
## Verifying this change
This change added tests and can be verified as follows:
-
`DefaultOperatorCoordinatorHandlerTest#testDisposeAllOperatorCoordinatorsClosesThoseOfEveryVertex`:
one registered vertex, one initialized but never registered, one
uninitialized; all three must be closed. Fails without the fix.
-
`AdaptiveBatchSchedulerITCase#testCancelInterruptsDynamicSourceParallelismInference`:
cancels a job whose source's inference blocks, then waits for the inference to
be interrupted. Hangs without the fix; passes in under a second with it.
- Ran alongside: all of `AdaptiveBatchSchedulerITCase` (7) and every test in
the `executiongraph`, `scheduler`, `operators.coordination` and
`source.coordinator` packages of flink-runtime (1,183 run, 2 skipped).
We have been running this fix in production on a 2.3.0-based build.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes (JobMaster shutdown
closes more operator coordinators)
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [ ] Yes (Claude-code 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]
