Utkarsh Vashishtha created FLINK-41090:
------------------------------------------
Summary: Operator coordinators of uninitialized vertices are never
closed
Key: FLINK-41090
URL: https://issues.apache.org/jira/browse/FLINK-41090
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Affects Versions: 2.3.0, 2.2.0, 2.1.0, 1.20.0, 1.19.0, 2.0.0
Reporter: Utkarsh Vashishtha
Since FLINK-33768, {{ExecutionJobVertex}} creates its operator coordinators in
its constructor, but
{{DefaultOperatorCoordinatorHandler#disposeAllOperatorCoordinators}} only
closes the ones in its {{coordinatorMap}}, i.e. those of initialized vertices.
With the {{AdaptiveBatchScheduler}}, a source vertex stays uninitialized while
its {{SourceCoordinator}} runs
{{DynamicParallelismInference#inferParallelism}}. If the job is cancelled or
fails during inference, the {{SourceCoordinator}} is never closed: the
inference isn't interrupted and runs to completion, and its
{{SourceCoordinator-<name>}} thread leaks.
*Repro:* a batch job whose source's {{inferParallelism}} blocks, cancelled
while it infers, on a standalone session cluster (and on a MiniCluster). The
job is CANCELED and the JobMaster stops within milliseconds, but the inference
runs to completion and its {{SourceCoordinator}} thread stays alive afterwards.
Affects every release since 1.19.0; 1.18 created coordinators on initialization
and had no source parallelism inference.
*Fix:* on dispose, close the coordinators of every vertex, not just the
registered ones. PR to follow.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
