You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2025/02/07 10:37:37 UTC
(flink) branch release-1.19 updated (af041a2a883 -> 216d3923380)
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a change to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
from af041a2a883 [FLINK-37205][python] Correct the state cache behavior during bump be… (#26059)
new eef910f7098 [FLINK-34227][runtime] Fixes concurrency issue in JobMaster shutdown (#24489)
new 216d3923380 [FLINK-34227][runtime] Fixes IO thread leaking into owner of the scheduler instance (i.e. the JobMaster)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/flink/util/concurrent/FutureUtils.java | 34 ++++++++-
.../apache/flink/runtime/jobmaster/JobMaster.java | 31 ++++++---
.../flink/runtime/scheduler/SchedulerBase.java | 5 +-
.../scheduler/adaptive/AdaptiveScheduler.java | 7 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 81 ++++++++++++++++++++++
.../slotpool/TestingSlotPoolServiceBuilder.java | 5 ++
.../runtime/scheduler/DefaultSchedulerTest.java | 70 +++++++++++++++++++
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 14 ++++
.../adaptivebatch/AdaptiveBatchSchedulerTest.java | 23 ++++++
9 files changed, 255 insertions(+), 15 deletions(-)
(flink) 02/02: [FLINK-34227][runtime] Fixes IO thread leaking into owner of the scheduler instance (i.e. the JobMaster)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 216d3923380ea91383337678cda82be233aa9bb6
Author: Matthias Pohl <mp...@confluent.io>
AuthorDate: Fri Jan 31 13:04:32 2025 +0100
[FLINK-34227][runtime] Fixes IO thread leaking into owner of the scheduler instance (i.e. the JobMaster)
- Adds test for checking whether the scheduler closing leaks an IO thread via the CheckpointsCleaner to the *SchedulerTests
- Makes CheckpointsCleaner available in AdaptiveBatchSchedulerFactory.createScheduler
---
.../apache/flink/util/concurrent/FutureUtils.java | 34 ++++++++++-
.../flink/runtime/scheduler/SchedulerBase.java | 5 +-
.../scheduler/adaptive/AdaptiveScheduler.java | 7 ++-
.../runtime/scheduler/DefaultSchedulerTest.java | 70 ++++++++++++++++++++++
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 14 +++++
.../adaptivebatch/AdaptiveBatchSchedulerTest.java | 23 +++++++
6 files changed, 146 insertions(+), 7 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java b/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java
index f20baa9f2cc..91e338f42c7 100644
--- a/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java
@@ -587,13 +587,41 @@ public class FutureUtils {
*/
public static CompletableFuture<Void> composeAfterwards(
CompletableFuture<?> future, Supplier<CompletableFuture<?>> composedAction) {
+ return composeAfterwardsInternal(future, composedAction, CompletableFuture::whenComplete);
+ }
+
+ /**
+ * Run the given asynchronous action after the completion of the given future. The given future
+ * can be completed normally or exceptionally. In case of an exceptional completion, the
+ * asynchronous action's exception will be added to the initial exception.
+ *
+ * @param future to wait for its completion
+ * @param composedAction asynchronous action which is triggered after the future's completion
+ * @return Future which is completed on the passed {@link Executor} after the asynchronous
+ * action has completed. This future can contain an exception if an error occurred in the
+ * given future or asynchronous action.
+ */
+ public static CompletableFuture<Void> composeAfterwardsAsync(
+ CompletableFuture<?> future,
+ Supplier<CompletableFuture<?>> composedAction,
+ Executor executor) {
+ return composeAfterwardsInternal(
+ future,
+ composedAction,
+ (composedActionFuture, resultFutureCompletion) ->
+ composedActionFuture.whenCompleteAsync(resultFutureCompletion, executor));
+ }
+
+ private static CompletableFuture<Void> composeAfterwardsInternal(
+ CompletableFuture<?> future,
+ Supplier<CompletableFuture<?>> composedAction,
+ BiConsumer<CompletableFuture<?>, BiConsumer<Object, Throwable>> forwardAction) {
final CompletableFuture<Void> resultFuture = new CompletableFuture<>();
future.whenComplete(
(Object outerIgnored, Throwable outerThrowable) -> {
- final CompletableFuture<?> composedActionFuture = composedAction.get();
-
- composedActionFuture.whenComplete(
+ forwardAction.accept(
+ composedAction.get(),
(Object innerIgnored, Throwable innerThrowable) -> {
if (innerThrowable != null) {
resultFuture.completeExceptionally(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
index 3c9241c6b41..31654223af9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
@@ -665,12 +665,13 @@ public abstract class SchedulerBase implements SchedulerNG, CheckpointScheduling
final FlinkException cause = new FlinkException("Scheduler is being stopped.");
final CompletableFuture<Void> checkpointServicesShutdownFuture =
- FutureUtils.composeAfterwards(
+ FutureUtils.composeAfterwardsAsync(
executionGraph
.getTerminationFuture()
.thenAcceptAsync(
this::shutDownCheckpointServices, getMainThreadExecutor()),
- checkpointsCleaner::closeAsync);
+ checkpointsCleaner::closeAsync,
+ getMainThreadExecutor());
FutureUtils.assertNoException(checkpointServicesShutdownFuture);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 238c594fd55..ff8f307ec68 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -571,12 +571,15 @@ public class AdaptiveScheduler
backgroundTask.abort();
// wait for the background task to finish and then close services
- return FutureUtils.composeAfterwards(
+ return FutureUtils.composeAfterwardsAsync(
FutureUtils.runAfterwardsAsync(
backgroundTask.getTerminationFuture(),
() -> stopCheckpointServicesSafely(jobTerminationFuture.get()),
getMainThreadExecutor()),
- checkpointsCleaner::closeAsync);
+ // closing the CheckpointsCleaner can complete in the ioExecutor when cleaning up a
+ // PendingCheckpoint
+ checkpointsCleaner::closeAsync,
+ getMainThreadExecutor());
}
private void stopCheckpointServicesSafely(JobStatus terminalState) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
index d109bb635d7..e9a3cc9cac5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
@@ -101,8 +101,10 @@ import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor;
import org.apache.flink.util.concurrent.ScheduledExecutor;
+import org.apache.flink.util.function.BiFunctionWithException;
import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
@@ -1786,6 +1788,19 @@ public class DefaultSchedulerTest {
}
}
+ @Test
+ void testCloseAsyncReturnsMainThreadFuture() throws Exception {
+ runCloseAsyncCompletesInMainThreadTest(
+ scheduledExecutorService,
+ (mainThreadExecutor, checkpointsCleaner) ->
+ createSchedulerBuilder(
+ singleJobVertexJobGraph(1),
+ mainThreadExecutor,
+ Collections.emptyList())
+ .setCheckpointCleaner(checkpointsCleaner)
+ .build());
+ }
+
@Test
void testJobStatusHookWithJobFailed() throws Exception {
commonJobStatusHookTest(ExecutionState.FAILED, JobStatus.FAILED);
@@ -1931,6 +1946,61 @@ public class DefaultSchedulerTest {
schedulerClosed.get();
}
+ // visible to expose test logic to other Scheduler test classes
+ public static void runCloseAsyncCompletesInMainThreadTest(
+ ScheduledExecutorService singleThreadExecutorService,
+ BiFunctionWithException<
+ ComponentMainThreadExecutor, CheckpointsCleaner, SchedulerNG, Exception>
+ schedulerFactory)
+ throws Exception {
+ final OneShotLatch cleanerCloseLatch = new OneShotLatch();
+ final CompletableFuture<Void> cleanerCloseFuture = new CompletableFuture<>();
+ final CheckpointsCleaner checkpointsCleaner =
+ new CheckpointsCleaner() {
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ cleanerCloseLatch.trigger();
+ return cleanerCloseFuture;
+ }
+ };
+
+ final ComponentMainThreadExecutor mainThreadExecutor =
+ ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+ singleThreadExecutorService);
+ final SchedulerNG scheduler =
+ schedulerFactory.apply(mainThreadExecutor, checkpointsCleaner);
+
+ mainThreadExecutor.execute(scheduler::startScheduling);
+
+ final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+ mainThreadExecutor.execute(
+ () -> {
+ // we shouldn't block the closeAsync call here because it's triggering
+ // additional tasks on the main thread internally
+ FutureUtils.forward(
+ scheduler
+ .closeAsync()
+ .thenRun(
+ () -> {
+ mainThreadExecutor.assertRunningInMainThread();
+ }),
+ closeFuture);
+ });
+
+ // wait for the CheckpointsCleaner#close call to not complete the future prematurely
+ cleanerCloseLatch.await();
+
+ // there is a race condition between returning the future and completing it which is due to
+ // the fact that we are triggering the latch before returning the future. That gives a small
+ // chance that the future completion is executed too early causing the future composition to
+ // end up in the main thread which is what we prevent in this test
+ Thread.sleep(50);
+ // completing this future in the test code simulates completing the
+ // CheckpointCleaner#closeAsync outside the main thread
+ cleanerCloseFuture.complete(null);
+ closeFuture.join();
+ }
+
private static long initiateFailure(
DefaultScheduler scheduler,
ExecutionAttemptID executionAttemptId,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
index 979d16a29a0..7e3476ddf78 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -141,6 +141,7 @@ import java.util.stream.Collectors;
import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createNoOpVertex;
+import static org.apache.flink.runtime.jobgraph.JobGraphTestUtils.singleNoOpJobGraph;
import static org.apache.flink.runtime.jobgraph.JobGraphTestUtils.streamingJobGraph;
import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements;
import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.offerSlots;
@@ -945,6 +946,19 @@ public class AdaptiveSchedulerTest {
assertThat(checkpointIdCounterShutdownFuture.get()).isEqualTo(JobStatus.FAILED);
}
+ @Test
+ void testCloseAsyncReturnsMainThreadFuture() throws Exception {
+ DefaultSchedulerTest.runCloseAsyncCompletesInMainThreadTest(
+ TEST_EXECUTOR_RESOURCE.getExecutor(),
+ (mainThreadExecutor, checkpointsCleaner) ->
+ new AdaptiveSchedulerBuilder(
+ singleNoOpJobGraph(),
+ mainThreadExecutor,
+ EXECUTOR_RESOURCE.getExecutor())
+ .setCheckpointCleaner(checkpointsCleaner)
+ .build());
+ }
+
@Test
void testTransitionToStateCallsOnLeave() throws Exception {
final AdaptiveScheduler scheduler =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java
index 5c18de43f54..b1a4cafce96 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java
@@ -68,11 +68,13 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import static org.apache.flink.runtime.scheduler.DefaultSchedulerBuilder.createCustomParallelismDecider;
+import static org.apache.flink.runtime.scheduler.DefaultSchedulerTest.runCloseAsyncCompletesInMainThreadTest;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createFailedTaskExecutionState;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createFinishedTaskExecutionState;
import static org.apache.flink.runtime.scheduler.adaptivebatch.DefaultVertexParallelismAndInputInfosDeciderTest.createDecider;
@@ -379,6 +381,22 @@ class AdaptiveBatchSchedulerTest {
assertThat(mergedSourceParallelismFuture.join()).isEqualTo(4);
}
+ @Test
+ void testCloseAsyncReturnsMainThreadFuture() throws Exception {
+ final ScheduledExecutorService scheduledExecutorServiceForMainThread =
+ Executors.newSingleThreadScheduledExecutor();
+ try {
+ runCloseAsyncCompletesInMainThreadTest(
+ scheduledExecutorServiceForMainThread,
+ (mainThread, checkpointsCleaner) ->
+ createSchedulerBuilder(createJobGraph(), mainThread)
+ .setCheckpointCleaner(checkpointsCleaner)
+ .buildAdaptiveBatchJobScheduler());
+ } finally {
+ scheduledExecutorServiceForMainThread.shutdownNow();
+ }
+ }
+
void testUserConfiguredMaxParallelism(
int globalMinParallelism,
int globalMaxParallelism,
@@ -567,6 +585,11 @@ class AdaptiveBatchSchedulerTest {
}
private DefaultSchedulerBuilder createSchedulerBuilder(JobGraph jobGraph) {
+ return createSchedulerBuilder(jobGraph, mainThreadExecutor);
+ }
+
+ private DefaultSchedulerBuilder createSchedulerBuilder(
+ JobGraph jobGraph, ComponentMainThreadExecutor mainThreadExecutor) {
return new DefaultSchedulerBuilder(
jobGraph, mainThreadExecutor, EXECUTOR_RESOURCE.getExecutor())
.setDelayExecutor(taskRestartExecutor);
(flink) 01/02: [FLINK-34227][runtime] Fixes concurrency issue in JobMaster shutdown (#24489)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit eef910f70988a12bb85022674763bbc35448ddf7
Author: Matthias Pohl <ma...@apache.org>
AuthorDate: Thu Feb 6 11:35:51 2025 +0100
[FLINK-34227][runtime] Fixes concurrency issue in JobMaster shutdown (#24489)
* Makes stopping the job execution in JobMaster run in the main thread
* Makes JobMaster#closeAsync more robust to disconnect calls during shutdown
* Adds test to cover FLINK-34227 scenario
---
.../apache/flink/runtime/jobmaster/JobMaster.java | 31 ++++++---
.../flink/runtime/jobmaster/JobMasterTest.java | 81 ++++++++++++++++++++++
.../slotpool/TestingSlotPoolServiceBuilder.java | 5 ++
3 files changed, 109 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 6edaa00ef07..2205e324677 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -839,14 +839,29 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
public void disconnectResourceManager(
final ResourceManagerId resourceManagerId, final Exception cause) {
- if (isConnectingToResourceManager(resourceManagerId)) {
+ if (resourceManagerAddress == null) {
+ log.debug(
+ "Disconnecting ResourceManager {} was triggered with no ResourceManager address "
+ + "being set (anymore). That either indicates that the ResourceManager "
+ + "lost leadership in the mean time or the message was received while "
+ + "shutting down the JobMaster. No reconnect will be initiated.",
+ resourceManagerId);
+ } else if (!resourceManagerAddress.getResourceManagerId().equals(resourceManagerId)) {
+ log.debug(
+ "Disconnecting ResourceManager {} was received while this instance is currently "
+ + "connected to another ResourceManager {} indicating that a ResourceManager "
+ + "leader change happened. No reconnect will be initiated.",
+ resourceManagerId,
+ resourceManagerAddress.getResourceManagerId());
+ } else {
reconnectToResourceManager(cause);
}
}
- private boolean isConnectingToResourceManager(ResourceManagerId resourceManagerId) {
- return resourceManagerAddress != null
- && resourceManagerAddress.getResourceManagerId().equals(resourceManagerId);
+ private void shutdownResourceManagerConnection(Exception cause) {
+ // unsetting the resourceManagerAddress will prevent reconnection
+ resourceManagerAddress = null;
+ closeResourceManagerConnection(cause);
}
@Override
@@ -1045,13 +1060,14 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
final CompletableFuture<Void> terminationFuture = stopScheduling();
- return FutureUtils.runAfterwards(
+ return FutureUtils.runAfterwardsAsync(
terminationFuture,
() -> {
shuffleMaster.unregisterJob(jobGraph.getJobID());
disconnectTaskManagerResourceManagerConnections(cause);
stopJobMasterServices();
- });
+ },
+ getMainThreadExecutor());
}
private void disconnectTaskManagerResourceManagerConnections(Exception cause) {
@@ -1063,8 +1079,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
disconnectTaskManager(taskManagerResourceId, cause);
}
- // disconnect from resource manager:
- closeResourceManagerConnection(cause);
+ shutdownResourceManagerConnection(cause);
}
private void stopHeartbeatServices() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index e8e63652283..2b5cc013f3b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -160,6 +160,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -2099,6 +2100,86 @@ class JobMasterTest {
}
}
+ @Test
+ public void testJobMasterDoesNotReconnectToResourceManagerEvenIfCleanupStalls()
+ throws Exception {
+
+ // the ResourceManager should count the connect attempts
+ final TestingResourceManagerGateway resourceManagerGateway =
+ createAndRegisterTestingResourceManagerGateway();
+ final AtomicInteger connectCount = new AtomicInteger();
+ final OneShotLatch firstRegistrationLatch = new OneShotLatch();
+ resourceManagerGateway.setRegisterJobManagerFunction(
+ (jobMasterId, resourceID, s, jobID) -> {
+ connectCount.incrementAndGet();
+ firstRegistrationLatch.trigger();
+
+ return CompletableFuture.completedFuture(
+ resourceManagerGateway.getJobMasterRegistrationSuccess());
+ });
+
+ final OneShotLatch schedulerCloseLatch = new OneShotLatch();
+ final CompletableFuture<Void> schedulerCloseFuture = new CompletableFuture<>();
+ final TestingSchedulerNG scheduler =
+ TestingSchedulerNG.newBuilder()
+ .setCloseAsyncSupplier(
+ () -> {
+ schedulerCloseLatch.trigger();
+ return schedulerCloseFuture;
+ })
+ .build();
+
+ final OneShotLatch closeSlotPoolLatch = new OneShotLatch();
+ final JobMaster jobMaster =
+ new JobMasterBuilder(jobGraph, rpcService)
+ .withHighAvailabilityServices(haServices)
+ .withSlotPoolServiceSchedulerFactory(
+ DefaultSlotPoolServiceSchedulerFactory.create(
+ TestingSlotPoolServiceBuilder.newBuilder()
+ .setCloseRunnable(closeSlotPoolLatch::awaitQuietly),
+ new TestingSchedulerNGFactory(scheduler)))
+ .createJobMaster();
+ jobMaster.start();
+
+ notifyResourceManagerLeaderListeners(resourceManagerGateway);
+ firstRegistrationLatch.await();
+
+ final CompletableFuture<Void> jobMasterCloseFuture = jobMaster.closeAsync();
+
+ // force the scheduler closing to happen outside the main thread
+ schedulerCloseLatch.await();
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ // we don't want the future to be completed right away because that could
+ // lead to the subsequent calls being executed by the main thread again
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ schedulerCloseFuture.complete(null);
+ });
+
+ // wait for the ResourceManager connection to be closed (happens before closing the slot
+ // pool)
+ CommonTestUtils.waitUntilCondition(() -> closeSlotPoolLatch.getWaitersCount() > 0);
+
+ final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class);
+ jobMasterGateway.disconnectResourceManager(
+ resourceManagerGateway.getFencingToken(),
+ new FlinkException(
+ "The JobMaster triggered a connection shutdown which is confirmed by the ResourceManager through this disconnectResourceManager call."));
+
+ // unblocks JobMaster's close procedure
+ closeSlotPoolLatch.trigger();
+ assertThatFuture(jobMasterCloseFuture).eventuallySucceeds();
+
+ assertThat(connectCount.get())
+ .as(
+ "The disconnect shouldn't trigger another reconnect because the ResourceManager connection was already closed.")
+ .isEqualTo(1);
+ }
+
@Test
void testRetrievingCheckpointStats() throws Exception {
// create savepoint data
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java
index aed1790d597..c99c8e91f12 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java
@@ -91,6 +91,11 @@ public class TestingSlotPoolServiceBuilder implements SlotPoolServiceFactory {
return this;
}
+ public TestingSlotPoolServiceBuilder setCloseRunnable(Runnable closeRunnable) {
+ this.closeRunnable = closeRunnable;
+ return this;
+ }
+
public static TestingSlotPoolServiceBuilder newBuilder() {
return new TestingSlotPoolServiceBuilder();
}