This is an automated email from the ASF dual-hosted git repository.
lihaosky 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 61723a08db5 [FLINK-37892][tests] Fix spurious TimeoutException in
TestUtils#waitUntil for non-monotonic conditions (#28400)
61723a08db5 is described below
commit 61723a08db52600363c89905913ce493a4f35d83
Author: Hao Li <[email protected]>
AuthorDate: Thu Jun 11 19:56:36 2026 -0700
[FLINK-37892][tests] Fix spurious TimeoutException in TestUtils#waitUntil
for non-monotonic conditions (#28400)
TestUtils#waitUntil evaluated the condition twice: once in the polling
loop and again to decide whether to throw. A condition that is
momentarily true and then false again (e.g. a live thread count matched
with ==) could exit the loop on the first evaluation and then fail the
second one, producing a TimeoutException within milliseconds despite a
30s budget. This is the failure mode behind
SplitFetcherManagerTest#testCloseBlockingWaitingForFetcherShutdown
flaking on CI with sub-second 'timeouts'.
Evaluate the condition once per iteration and latch the result, so a
condition observed as true always completes the wait. Add a regression
test with a condition that is satisfied exactly once.
---
.../java/org/apache/flink/test/util/TestUtils.java | 15 ++++--
.../org/apache/flink/test/util/TestUtilsTest.java | 60 ++++++++++++++++++++++
2 files changed, 70 insertions(+), 5 deletions(-)
diff --git
a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
index e3780a7eeaf..fc424bbd03a 100644
---
a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
+++
b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
@@ -223,13 +223,18 @@ public class TestUtils {
*/
public static void waitUntil(Supplier<Boolean> condition, Duration
timeout, String message)
throws InterruptedException, TimeoutException {
- long startTime = System.currentTimeMillis();
- while (!condition.get() && System.currentTimeMillis() < startTime +
timeout.toMillis()) {
+ long deadline = System.currentTimeMillis() + timeout.toMillis();
+ while (true) {
+ // Latch the result so that a condition that is satisfied once
does not fail the
+ // wait if it becomes false again before a re-evaluation.
+ if (condition.get()) {
+ return;
+ }
+ if (System.currentTimeMillis() >= deadline) {
+ throw new TimeoutException(message);
+ }
Thread.sleep(1);
}
- if (!condition.get()) {
- throw new TimeoutException(message);
- }
}
private static boolean allVerticesRunning(Map<ExecutionState, Integer>
states) {
diff --git
a/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/test/util/TestUtilsTest.java
b/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/test/util/TestUtilsTest.java
new file mode 100644
index 00000000000..bcc42e5a8fc
--- /dev/null
+++
b/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/test/util/TestUtilsTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TestUtils}. */
+class TestUtilsTest {
+
+ @Test
+ void testWaitUntilReturnsWhenConditionIsMet() {
+ assertThatCode(() -> TestUtils.waitUntil(() -> true,
Duration.ofSeconds(5), "msg"))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testWaitUntilThrowsOnTimeout() {
+ assertThatThrownBy(() -> TestUtils.waitUntil(() -> false,
Duration.ofMillis(50), "msg"))
+ .isInstanceOf(TimeoutException.class)
+ .hasMessage("msg");
+ }
+
+ @Test
+ void testWaitUntilSucceedsForConditionSatisfiedOnlyOnce() {
+ // A non-monotonic condition may become true and then false again.
Once it has been
+ // observed as true, waitUntil must return successfully instead of
re-evaluating it
+ // and throwing a spurious TimeoutException.
+ final AtomicBoolean firstCall = new AtomicBoolean(true);
+ assertThatCode(
+ () ->
+ TestUtils.waitUntil(
+ () -> firstCall.getAndSet(false),
+ Duration.ofSeconds(5),
+ "msg"))
+ .doesNotThrowAnyException();
+ }
+}
