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 4902753429f [FLINK-35562][table-planner] Bound result wait in
RestoreTestBase to avoid CI hangs (#28404)
4902753429f is described below
commit 4902753429fcca8975de66843bb3eab2b0e5a0de
Author: Martijn Visser <[email protected]>
AuthorDate: Thu Jun 18 06:56:47 2026 -0700
[FLINK-35562][table-planner] Bound result wait in RestoreTestBase to avoid
CI hangs (#28404)
* [FLINK-35562][table-planner] Bound result wait in RestoreTestBase to
avoid CI hangs
testRestore waits on an unbounded future that never completes when a
PROCTIME() window splits records differently than when the expected data
was captured, hanging the surefire fork until the CI watchdog kills it.
Bound the wait at 5 minutes and fail with the program id and the actual
sink results. Only the two proc-time restore tests (FLINK-35562,
FLINK-34404) use the affected INFINITE path; the proc-time nondeterminism
itself is left as follow-up.
Generated-by: Claude Opus 4.8 (1M context)
---
.../plan/nodes/exec/testutils/RestoreTestBase.java | 52 ++++++++++++++++++++--
1 file changed, 49 insertions(+), 3 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
index d64ac4f858e..f1cdc2ca0bc 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
@@ -76,6 +76,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -106,6 +108,10 @@ public abstract class RestoreTestBase implements
TableTestProgramRunner {
// This version can be set to generate savepoints for a particular Flink
version.
// By default, the savepoint is generated for the current version in the
default directory.
private static final FlinkVersion FLINK_VERSION = null;
+
+ // Bounds the wait for the expected sink results so a stuck job (e.g. an
unlucky
+ // processing-time window split) fails the test instead of hanging the
fork.
+ private static final long RESULT_AWAIT_TIMEOUT_MILLIS =
TimeUnit.MINUTES.toMillis(5);
private final Class<? extends ExecNode<?>> execNodeUnderTest;
private final List<Class<? extends ExecNode<?>>> childExecNodesUnderTest;
private final AfterRestoreSource afterRestoreSource;
@@ -276,6 +282,43 @@ public abstract class RestoreTestBase implements
TableTestProgramRunner {
});
}
+ private void awaitExpectedResults(
+ final TableTestProgram program,
+ final List<CompletableFuture<?>> futures,
+ final boolean ignoreAfter)
+ throws Exception {
+ try {
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .get(RESULT_AWAIT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ final StringBuilder message = new StringBuilder();
+ message.append("Sink did not produce the expected results within ")
+ .append(RESULT_AWAIT_TIMEOUT_MILLIS)
+ .append(" ms for program '")
+ .append(program.id)
+ .append("'.");
+ for (SinkTestStep sinkTestStep : program.getSetupSinkTestSteps()) {
+ // Mirror the expected set the observer waits on, so the dump
is exact.
+ final List<String> expected =
+ new
ArrayList<>(sinkTestStep.getExpectedBeforeRestoreAsStrings());
+ if (!ignoreAfter) {
+
expected.addAll(sinkTestStep.getExpectedAfterRestoreAsStrings());
+ }
+ message.append(System.lineSeparator())
+ .append("Sink '")
+ .append(sinkTestStep.name)
+ .append("' expected results: ")
+ .append(expected)
+ .append(System.lineSeparator())
+ .append("Sink '")
+ .append(sinkTestStep.name)
+ .append("' actual results: ")
+ .append(getActualResults(sinkTestStep,
sinkTestStep.name));
+ }
+ throw new AssertionError(message.toString(), e);
+ }
+ }
+
/**
* Execute this test to generate test files. Remember to be using the
correct branch when
* generating the test files.
@@ -336,7 +379,7 @@ public abstract class RestoreTestBase implements
TableTestProgramRunner {
compiledPlan.writeToFile(getPlanPath(program, getLatestMetadata()));
final TableResult tableResult = compiledPlan.execute();
- CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).get();
+ awaitExpectedResults(program, futures, true);
final JobClient jobClient = tableResult.getJobClient().get();
final String savepoint =
jobClient
@@ -431,8 +474,11 @@ public abstract class RestoreTestBase implements
TableTestProgramRunner {
if (afterRestoreSource == AfterRestoreSource.INFINITE) {
final TableResult tableResult = compiledPlan.execute();
- CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).get();
- tableResult.getJobClient().get().cancel().get();
+ try {
+ awaitExpectedResults(program, futures, false);
+ } finally {
+ tableResult.getJobClient().get().cancel().get();
+ }
} else {
compiledPlan.execute().await();
for (SinkTestStep sinkTestStep : program.getSetupSinkTestSteps()) {
