This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new f237727ed52 [FLINK-34427][tests] Wait for slot manager close() in
runTest
f237727ed52 is described below
commit f237727ed52dbe967bc022268b173fb68587b81b
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 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);
+ }
}
}
}
