[
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)
