This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 3df000c936d [FLINK-41090][runtime] Close the operator coordinators of
uninitialized vertices on dispose
3df000c936d is described below
commit 3df000c936d535690095354e07f6f306ab37677d
Author: Utkarsh Vashishtha <[email protected]>
AuthorDate: Fri Oct 23 14:16:15 2026 -0700
[FLINK-41090][runtime] Close the operator coordinators of uninitialized
vertices on dispose
Since FLINK-33768, ExecutionJobVertex creates its operator coordinators in
its constructor, but DefaultOperatorCoordinatorHandler only tracks those of
initialized vertices, so disposeAllOperatorCoordinators never closed the
coordinators of vertices that were not initialized yet.
With the AdaptiveBatchScheduler, a source vertex stays uninitialized while
its SourceCoordinator runs DynamicParallelismInference#inferParallelism. A
job cancelled or failed during inference therefore never closed that
coordinator: the inference was not interrupted and ran to completion, and
its coordinator thread leaked.
Close the coordinators of every vertex in the graph instead. This also
covers a vertex whose initialization failed after it counted as initialized
but before its coordinators were registered.
Generated-by: Claude Code (Claude Opus 5.5)
---
.../runtime/executiongraph/ExecutionJobVertex.java | 4 ++
.../DefaultOperatorCoordinatorHandler.java | 4 +-
.../DefaultOperatorCoordinatorHandlerTest.java | 39 ++++++++++++++++++
.../scheduling/AdaptiveBatchSchedulerITCase.java | 46 ++++++++++++++++++++++
4 files changed, 92 insertions(+), 1 deletion(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
index 325b931f6e5..f3edcd191ba 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
@@ -430,6 +430,10 @@ public class ExecutionJobVertex
return operatorCoordinators;
}
+ public Collection<OperatorCoordinatorHolder>
getOperatorCoordinatorsEvenIfUninitialized() {
+ return operatorCoordinators;
+ }
+
public List<SourceCoordinator<?, ?>> getSourceCoordinators() {
List<SourceCoordinator<?, ?>> sourceCoordinators = new ArrayList<>();
for (OperatorCoordinatorHolder oph : operatorCoordinators) {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
index 509118244c3..026d325136e 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandler.java
@@ -87,7 +87,9 @@ public class DefaultOperatorCoordinatorHandler implements
OperatorCoordinatorHan
@Override
public void disposeAllOperatorCoordinators() {
- coordinatorMap.values().forEach(IOUtils::closeQuietly);
+ executionGraph.getAllVertices().values().stream()
+ .flatMap(vertex ->
vertex.getOperatorCoordinatorsEvenIfUninitialized().stream())
+ .forEach(IOUtils::closeQuietly);
}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandlerTest.java
index 6f7c1625edd..4286731ed6b 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultOperatorCoordinatorHandlerTest.java
@@ -78,6 +78,45 @@ class DefaultOperatorCoordinatorHandlerTest {
assertThat(handler.getCoordinatorMap().keySet()).contains(operatorId1,
operatorId2);
}
+ @Test
+ void testDisposeAllOperatorCoordinatorsClosesThoseOfEveryVertex() throws
Exception {
+ final JobVertex[] jobVertices = createJobVertices(BLOCKING);
+ final JobVertex unconnected = createNoOpVertex(3);
+ unconnected.addOperatorCoordinator(
+ new SerializedValue<>(
+ new TestingOperatorCoordinator.Provider(
+
OperatorID.fromJobVertexID(unconnected.getID()))));
+
+ final ExecutionGraph executionGraph =
+ createDynamicGraph(jobVertices[0], jobVertices[1],
unconnected);
+ final ExecutionJobVertex registered =
executionGraph.getJobVertex(jobVertices[0].getID());
+ final ExecutionJobVertex neverRegistered =
+ executionGraph.getJobVertex(jobVertices[1].getID());
+ final ExecutionJobVertex uninitialized =
executionGraph.getJobVertex(unconnected.getID());
+
executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());
+ executionGraph.initializeJobVertex(registered, 0L);
+ final DefaultOperatorCoordinatorHandler handler =
+ new DefaultOperatorCoordinatorHandler(executionGraph,
throwable -> {});
+ executionGraph.initializeJobVertex(neverRegistered, 0L);
+
+ handler.disposeAllOperatorCoordinators();
+
+ assertClosed(registered);
+ assertClosed(neverRegistered);
+ assertClosed(uninitialized);
+ }
+
+ private static void assertClosed(ExecutionJobVertex vertex) {
+ assertThat(vertex.getOperatorCoordinatorsEvenIfUninitialized())
+ .singleElement()
+ .satisfies(
+ holder ->
+ assertThat(
+ ((TestingOperatorCoordinator)
holder.coordinator())
+ .isClosed())
+ .isTrue());
+ }
+
private JobVertex[] createJobVertices(ResultPartitionType
resultPartitionType)
throws IOException {
final JobVertex[] jobVertices = new JobVertex[2];
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
index adad853102e..765a965dd01 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
@@ -30,6 +30,8 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler;
import org.apache.flink.runtime.scheduler.adaptivebatch.OperatorsFinished;
@@ -37,6 +39,7 @@ import
org.apache.flink.runtime.scheduler.adaptivebatch.StreamGraphOptimizationS
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamGraphContext;
import org.apache.flink.streaming.api.graph.StreamNode;
@@ -128,6 +131,26 @@ class AdaptiveBatchSchedulerITCase {
testSchedulingBase(true);
}
+ @Test
+ void testCancelInterruptsDynamicSourceParallelismInference() throws
Exception {
+ BlockingParallelismInferenceSource.INFERRING.reset();
+ BlockingParallelismInferenceSource.INTERRUPTED.reset();
+
+ final StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.createLocalEnvironment(createConfiguration());
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ env.fromSource(
+ new BlockingParallelismInferenceSource(),
+ WatermarkStrategy.noWatermarks(),
+ "source")
+ .sinkTo(new DiscardingSink<>());
+
+ final JobClient jobClient = env.executeAsync();
+ BlockingParallelismInferenceSource.INFERRING.await();
+ jobClient.cancel().get();
+ BlockingParallelismInferenceSource.INTERRUPTED.await();
+ }
+
@Test
void testParallelismOfForwardGroupLargerThanGlobalMaxParallelism() throws
Exception {
final Configuration configuration = createConfiguration();
@@ -371,6 +394,29 @@ class AdaptiveBatchSchedulerITCase {
}
}
+ private static class BlockingParallelismInferenceSource extends
NumberSequenceSource
+ implements DynamicParallelismInference {
+ private static final long serialVersionUID = 1L;
+ private static final OneShotLatch INFERRING = new OneShotLatch();
+ private static final OneShotLatch INTERRUPTED = new OneShotLatch();
+
+ BlockingParallelismInferenceSource() {
+ super(0, 1);
+ }
+
+ @Override
+ public int inferParallelism(Context context) {
+ INFERRING.trigger();
+ try {
+ new OneShotLatch().await();
+ } catch (InterruptedException e) {
+ INTERRUPTED.trigger();
+ Thread.currentThread().interrupt();
+ }
+ return 1;
+ }
+ }
+
public static final class TestingStreamGraphOptimizerStrategy
implements StreamGraphOptimizationStrategy {
