This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f2aa53456cb [FLINK-34427][tests] Wait for slot manager close() in
runTest
f2aa53456cb is described below
commit f2aa53456cb99e535b217d555bb7af2e01a3f744
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)
---
.../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 0edd01247f7..b9799bf1588 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
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
import org.apache.flink.api.common.JobID;
-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;
@@ -36,6 +35,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);
+ }
}
}
}
