This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 4048a892875 [FLINK-40070][tests] Fix flaky DynamicParameterITCase
reading rolled JobManager logs
4048a892875 is described below
commit 4048a8928756f1bf187400f8e5aa4af5298fdd3e
Author: Martijn Visser <[email protected]>
AuthorDate: Tue Jun 16 11:30:28 2026 -0700
[FLINK-40070][tests] Fix flaky DynamicParameterITCase reading rolled
JobManager logs
The distribution log4j configuration rolls the log file on startup, so the
JobManager startup banner frequently lands in a rolled .log.N file that
FlinkDistribution.searchAllLogs skips; the test then either spins unboundedly
waiting for the banner (multi-hour e2e_4 hang) or parses a half-written
arguments block ("Missing required option: c"). Search rolled logs for the
startup banner and bound the wait so a missing banner fails fast.
Generated-by: Claude Opus 4.8 (1M context)
(cherry picked from commit f9e0756fbfd15c633ae416c2e76f634da46101e8)
---
.../flink/tests/util/flink/FlinkDistribution.java | 17 ++++++-
.../apache/flink/dist/DynamicParameterITCase.java | 54 ++++++++++++++++++----
2 files changed, 60 insertions(+), 11 deletions(-)
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
index 404b1ed4555..f4192a53d01 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
@@ -435,13 +435,28 @@ public final class FlinkDistribution {
public <T> Stream<T> searchAllLogs(Pattern pattern, Function<Matcher, T>
matchProcessor)
throws IOException {
+ return searchAllLogs(pattern, matchProcessor, false);
+ }
+
+ /**
+ * Searches the distribution's log files for lines matching the given
pattern.
+ *
+ * @param includeRolledLogs whether to also search rolled log files
({@code .log.N}); the
+ * distribution rolls logs on startup, which can move the startup
banner out of the live
+ * {@code .log} file
+ */
+ public <T> Stream<T> searchAllLogs(
+ Pattern pattern, Function<Matcher, T> matchProcessor, boolean
includeRolledLogs)
+ throws IOException {
final List<T> matches = new ArrayList<>(2);
try (Stream<Path> logFilesStream = Files.list(log)) {
final Iterator<Path> logFiles = logFilesStream.iterator();
while (logFiles.hasNext()) {
final Path logFile = logFiles.next();
- if (!logFile.getFileName().toString().endsWith(".log")) {
+ final String fileName = logFile.getFileName().toString();
+ final boolean isRolledLog =
fileName.matches(".*\\.log\\.\\d+");
+ if (!fileName.endsWith(".log") && !(includeRolledLogs &&
isRolledLog)) {
// ignore logs for previous runs that have a number suffix
continue;
}
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/test/java/org/apache/flink/dist/DynamicParameterITCase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/test/java/org/apache/flink/dist/DynamicParameterITCase.java
index e34c471eb1a..7d8a685744b 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/test/java/org/apache/flink/dist/DynamicParameterITCase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/test/java/org/apache/flink/dist/DynamicParameterITCase.java
@@ -17,6 +17,7 @@
package org.apache.flink.dist;
+import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.entrypoint.EntrypointClusterConfiguration;
import org.apache.flink.runtime.entrypoint.FlinkParseException;
@@ -31,7 +32,9 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.nio.file.Path;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Predicate;
@@ -47,6 +50,8 @@ class DynamicParameterITCase {
private static final Pattern ENTRYPOINT_CLASSPATH_LOG_PATTERN =
Pattern.compile(".*ClusterEntrypoint +\\[] - +Classpath:.*");
+ private static final Duration LOG_TIMEOUT = Duration.ofMinutes(1);
+
private static final String HOST = "localhost";
private static final int PORT = 8081;
@@ -122,16 +127,14 @@ class DynamicParameterITCase {
dist.callJobManagerScript(args.toArray(new String[0]));
- while (!allProgramArgumentsLogged(dist)) {
- Thread.sleep(500);
- }
-
- try (Stream<String> lines =
- dist.searchAllLogs(ENTRYPOINT_LOG_PATTERN, matcher ->
matcher.group(1))) {
+ // The backgrounded JobManager writes its log asynchronously; wait
until the complete
+ // program arguments block has been flushed, otherwise the parser may
observe a
+ // partially-written block or never observe it at all.
+ final String[] programArguments = waitForProgramArguments(dist);
+ try {
final EntrypointClusterConfiguration entrypointConfig =
- ClusterEntrypoint.parseArguments(
- lines.filter(new
ProgramArgumentsFilter()).toArray(String[]::new));
+ ClusterEntrypoint.parseArguments(programArguments);
assertThat(entrypointConfig.getHostname()).isEqualTo(withHost ?
HOST : null);
assertThat(entrypointConfig.getRestPort()).isEqualTo(withPort ?
PORT : -1);
@@ -148,11 +151,42 @@ class DynamicParameterITCase {
}
}
- private static boolean allProgramArgumentsLogged(FlinkDistribution dist)
throws IOException {
+ private static String[] waitForProgramArguments(FlinkDistribution dist)
throws Exception {
+ final List<String[]> captured = new ArrayList<>(1);
+ CommonTestUtils.waitUtil(
+ () -> {
+ // The "Classpath:" line is logged after the program
arguments, so its presence
+ // means the complete arguments block has been flushed.
+ if (!allProgramArgumentsLogged(dist)) {
+ return false;
+ }
+ captured.clear();
+ captured.add(readProgramArguments(dist));
+ return true;
+ },
+ LOG_TIMEOUT,
+ Duration.ofMillis(500),
+ "The JobManager did not log its complete program arguments in
time.");
+ return captured.get(0);
+ }
+
+ private static String[] readProgramArguments(FlinkDistribution dist) {
+ try (Stream<String> lines =
+ dist.searchAllLogs(ENTRYPOINT_LOG_PATTERN, matcher ->
matcher.group(1), true)) {
+ return lines.filter(new
ProgramArgumentsFilter()).toArray(String[]::new);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private static boolean allProgramArgumentsLogged(FlinkDistribution dist) {
// the classpath is logged after the program arguments
try (Stream<String> lines =
- dist.searchAllLogs(ENTRYPOINT_CLASSPATH_LOG_PATTERN, matcher
-> matcher.group(0))) {
+ dist.searchAllLogs(
+ ENTRYPOINT_CLASSPATH_LOG_PATTERN, matcher ->
matcher.group(0), true)) {
return lines.iterator().hasNext();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
}
}
