You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Utkarsh Vashishtha (Jira)" <ji...@apache.org> on 2026/10/22 21:22:00 UTC
[jira] [Updated] (FLINK-41090) Operator coordinators of uninitialized vertices are never closed
[ https://issues.apache.org/jira/browse/FLINK-41090?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Utkarsh Vashishtha updated FLINK-41090:
---------------------------------------
Description:
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. We run this fix in our production environment, PR to commit to upstream will follow.
was:
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.
> 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.0.0, 1.19.0, 1.20.0, 2.1.0, 2.2.0, 2.3.0
> Reporter: Utkarsh Vashishtha
> Priority: Major
>
> 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. We run this fix in our production environment, PR to commit to upstream will follow.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)