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
commit 1268adfdb03ff9dca2ce53a224974db6918b66d5 Author: Martijn Visser <[email protected]> AuthorDate: Mon Jun 29 10:56:37 2026 +0200 [FLINK-40009][tests] Enforce timeout in RescaleTimelineITCase wait helper waitUntilConditionWithTimeout wrapped an unbounded CommonTestUtils#waitUntilCondition in CompletableFuture#runAsync and bounded it with get(timeout). On JDK 11, when the test runs on a ForkJoinPool worker (JUnit 5's test executor), CompletableFuture#timedGet help-executes the async task inline via ForkJoinPool#helpAsyncBlocker, so the never-ending poll loop runs on the waiting thread and the timeout never fires. A CI thread dump of the JDK 11 cron leg showed the call still parked after 978s on a 20s budget, hanging the fork until the watchdog killed it. JDK 17 and 21 changed helpAsyncBlocker and do not hang. Poll synchronously on the calling thread via CommonTestUtils#waitUtil so the timeout is always enforced and no task is leaked into the common pool. Generated-by: Claude Opus 4.8 (1M context) --- .../adaptive/timeline/RescaleTimelineITCase.java | 28 +++++++++++++--------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java index 11711208453..cbf32d7467e 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java @@ -57,9 +57,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -77,6 +75,8 @@ class RescaleTimelineITCase { private static final int NUMBER_TASK_MANAGERS = 2; private static final int PARALLELISM = NUMBER_SLOTS_PER_TASK_MANAGER * NUMBER_TASK_MANAGERS; private static final JobVertexID JOB_VERTEX_ID = new JobVertexID(); + // Matches the default poll interval of the CommonTestUtils#waitUntilCondition it replaced. + private static final long RETRY_INTERVAL_MILLIS = 100L; @Parameter private Configuration configuration; private MiniClusterResource miniClusterResource; @@ -700,15 +700,21 @@ class RescaleTimelineITCase { private void waitUntilConditionWithTimeout( SupplierWithException<Boolean, Exception> condition, long timeoutMillis) throws Exception { - CompletableFuture.runAsync( - () -> { - try { - CommonTestUtils.waitUntilCondition(condition); - } catch (Exception e) { - throw new RuntimeException(e); - } - }) - .get(timeoutMillis, TimeUnit.MILLISECONDS); + // Poll synchronously via waitUtil. The previous CompletableFuture#runAsync + get(timeout) + // wrapper hangs on JDK 11: on a ForkJoinPool worker (JUnit's executor) timedGet + // help-executes the unbounded poll loop inline on the waiting thread, so the timeout + // never fires. + org.apache.flink.core.testutils.CommonTestUtils.waitUtil( + () -> { + try { + return condition.get(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }, + Duration.ofMillis(timeoutMillis), + Duration.ofMillis(RETRY_INTERVAL_MILLIS), + "Condition was not met within " + timeoutMillis + " ms."); } private void waitForVertexParallelismReachedAndJobRunning(
