This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new e00a10ee84a [FLINK-41090][runtime] Close the operator coordinators of
uninitialized vertices on dispose
e00a10ee84a is described below
commit e00a10ee84a0985d8663fcf61101aceea48f0335
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 f9c77f19e94..256961bec4f 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
@@ -411,6 +411,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 7896e97fbfa..e4133a45d85 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 e488787f5a1..40fbdc88801 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
@@ -77,6 +77,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 dd2afe6bc89..82fa6725e8b 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
@@ -31,9 +31,12 @@ 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.scheduler.adaptivebatch.AdaptiveBatchScheduler;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -82,6 +85,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();
@@ -259,4 +282,27 @@ class AdaptiveBatchSchedulerITCase {
return expectedParallelism;
}
}
+
+ 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;
+ }
+ }
}
