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 6a668bd419e [FLINK-34427][tests] Wait for slot manager close() in
runTest
6a668bd419e is described below
commit 6a668bd419e242ab0da5e68aac352dd9b442e722
Author: Martijn Visser <[email protected]>
AuthorDate: Tue Oct 20 18:10:39 2026 +0200
[FLINK-34427][tests] Wait for slot manager close() in runTest
runTest queued close() on the main thread executor and returned without
waiting for it. If a requirements check from the last test of the class
was still running, afterAll shut the executor down, the handleAsync in
DefaultSlotStatusSyncer#allocateSlot was rejected and
FatalExitExceptionHandler exited the fork with code 239.
Waiting for close() also waits for that check, and later checks return
early on the stopped slot manager. close() now also runs after a
failing test body, and close() and runInMainThreadAndWait go through
FutureUtils#runAsync, so a failure in either, Errors included, fails the
test instead of exiting or hanging the fork.
Generated-by: Claude Code (Claude Opus 5.5, Claude Fable 5.1)
(cherry picked from commit f2aa53456cb99e535b217d555bb7af2e01a3f744)
---
.../FineGrainedSlotManagerTestBase.java | 43 +++++++++++-----------
1 file changed, 21 insertions(+), 22 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTestBase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTestBase.java
index 23c6c89cf04..904411667a6 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTestBase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTestBase.java
@@ -19,7 +19,6 @@ package org.apache.flink.runtime.resourcemanager.slotmanager;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
-import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.blocklist.BlockedTaskManagerChecker;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -37,6 +36,7 @@ import
org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGateway;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.testutils.executor.TestExecutorExtension;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.concurrent.ScheduledExecutor;
@@ -49,6 +49,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -201,14 +202,9 @@ abstract class FineGrainedSlotManagerTestBase {
mainThreadExecutor.execute(runnable);
}
- void runInMainThreadAndWait(Runnable runnable) throws
InterruptedException {
- final OneShotLatch latch = new OneShotLatch();
- mainThreadExecutor.execute(
- () -> {
- runnable.run();
- latch.trigger();
- });
- latch.await();
+ void runInMainThreadAndWait(RunnableWithException runnable)
+ throws InterruptedException, ExecutionException {
+ FutureUtils.runAsync(runnable, mainThreadExecutor).get();
}
protected final void runTest(RunnableWithException testMethod) throws
Exception {
@@ -234,19 +230,22 @@ abstract class FineGrainedSlotManagerTestBase {
resourceEventListenerBuilder.build(),
blockedTaskManagerChecker));
- testMethod.run();
-
- CompletableFuture<Void> closeFuture = new CompletableFuture<>();
- runInMainThread(
- () -> {
- try {
- slotManager.close();
- } catch (Exception e) {
- closeFuture.completeExceptionally(e);
- }
- closeFuture.complete(null);
- });
- FutureUtils.assertNoException(closeFuture);
+ Throwable failure = null;
+ try {
+ testMethod.run();
+ } catch (Throwable t) {
+ failure = t;
+ }
+ // Wait for close(), so afterAll cannot shut the executor down
under a running
+ // requirements check. A failing close() does not hide the test's
own failure.
+ try {
+ runInMainThreadAndWait(slotManager::close);
+ } catch (Throwable t) {
+ failure = ExceptionUtils.firstOrSuppressed(t, failure);
+ }
+ if (failure != null) {
+ ExceptionUtils.rethrowException(failure);
+ }
}
}
}
