From e0b8843c7756b5cf7f850db0fe0a61075e0d25fd Mon Sep 17 00:00:00 2001 From: aIbrahiim Date: Thu, 9 Jul 2026 16:18:21 +0300 Subject: [PATCH 1/4] fix Flink runner tests on Java 21 --- .../trigger_files/beam_PreCommit_Java.json | 2 +- runners/flink/flink_runner.gradle | 17 ++++++ .../runners/flink/FlinkSubmissionTest.java | 54 ++++++++++++++----- 3 files changed, 60 insertions(+), 13 deletions(-) diff --git a/.github/trigger_files/beam_PreCommit_Java.json b/.github/trigger_files/beam_PreCommit_Java.json index 5abe02fc09c7..e797011e0364 100644 --- a/.github/trigger_files/beam_PreCommit_Java.json +++ b/.github/trigger_files/beam_PreCommit_Java.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 1 + "modification": 7 } diff --git a/runners/flink/flink_runner.gradle b/runners/flink/flink_runner.gradle index 837561ec71b7..165d5c2d93d7 100644 --- a/runners/flink/flink_runner.gradle +++ b/runners/flink/flink_runner.gradle @@ -180,6 +180,22 @@ if (use_override) { } } +def flinkTestJvmArgs() { + def testJavaVer = project.findProperty('testJavaVersion') ? (project.property('testJavaVersion') as int) : JavaVersion.current().majorVersion.toInteger() + if (testJavaVer >= 17) { + return [ + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/java.util=ALL-UNNAMED", + "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED", + "--add-opens=java.base/java.lang=ALL-UNNAMED", + "-Djava.security.manager=allow", + ] + } else { + return [] + } +} + test { systemProperty "log4j.configuration", "log4j-test.properties" // Change log level to debug: @@ -187,6 +203,7 @@ test { // Change log level to debug only for the package and nested packages: // systemProperty "org.slf4j.simpleLogger.log.org.apache.beam.runners.flink.translation.wrappers.streaming", "debug" jvmArgs "-XX:-UseGCOverheadLimit" + jvmArgs += flinkTestJvmArgs() if (System.getProperty("beamSurefireArgline")) { jvmArgs System.getProperty("beamSurefireArgline") } diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java index 8e4c3255fac5..538db8472d4a 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java @@ -19,11 +19,12 @@ import java.io.File; import java.lang.reflect.Field; -import java.lang.reflect.Modifier; +import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.security.Permission; import java.util.Collection; +import java.util.Locale; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -230,21 +231,50 @@ private static void restoreEnvironment() throws Exception { * We modify the JVM's environment variables here. This is necessary for the end-to-end test * because Flink's CliFrontend requires a Flink configuration file for which the location can only * be set using the {@code ConfigConstants.ENV_FLINK_CONF_DIR} environment variable. + * + *

On Unix JDKs, {@code ProcessEnvironment.theEnvironment} is a {@code Map}. + * {@code System.getenv(String)} looks up {@code Variable} keys, so inserting plain {@code String} + * keys (via {@code putAll}) is a no-op for lookups and leaves Flink unable to find {@code + * FLINK_CONF_DIR}. On Windows the map uses {@code String} keys (plus a case-insensitive map). */ + @SuppressWarnings("unchecked") private static void modifyEnv(Map env) throws Exception { - Class processEnv = Class.forName("java.lang.ProcessEnvironment"); - Field envField = processEnv.getDeclaredField("theUnmodifiableEnvironment"); + Class processEnvironment = Class.forName("java.lang.ProcessEnvironment"); - Field modifiersField = Field.class.getDeclaredField("modifiers"); - modifiersField.setAccessible(true); - modifiersField.setInt(envField, envField.getModifiers() & ~Modifier.FINAL); - - envField.setAccessible(true); - envField.set(null, env); - envField.setAccessible(false); + Field theEnvironmentField = processEnvironment.getDeclaredField("theEnvironment"); + theEnvironmentField.setAccessible(true); + Map envMap = (Map) theEnvironmentField.get(null); + envMap.clear(); + try { + // Unix: keys/values are ProcessEnvironment$Variable / $Value, not String. + Class variableClass = Class.forName("java.lang.ProcessEnvironment$Variable"); + Class valueClass = Class.forName("java.lang.ProcessEnvironment$Value"); + Method valueOfVariable = variableClass.getDeclaredMethod("valueOf", String.class); + Method valueOfValue = valueClass.getDeclaredMethod("valueOf", String.class); + valueOfVariable.setAccessible(true); + valueOfValue.setAccessible(true); + for (Map.Entry entry : env.entrySet()) { + envMap.put( + valueOfVariable.invoke(null, entry.getKey()), + valueOfValue.invoke(null, entry.getValue())); + } + } catch (ClassNotFoundException e) { + // Windows (and similar): theEnvironment uses String keys. + envMap.putAll(env); + } - modifiersField.setInt(envField, envField.getModifiers() & Modifier.FINAL); - modifiersField.setAccessible(false); + // Windows: System.getenv(String) reads the case-insensitive map. + if (System.getProperty("os.name", "").toLowerCase(Locale.ROOT).startsWith("windows")) { + Field theCaseInsensitiveEnvironmentField = + processEnvironment.getDeclaredField("theCaseInsensitiveEnvironment"); + theCaseInsensitiveEnvironmentField.setAccessible(true); + Map ciEnvMap = + (Map) theCaseInsensitiveEnvironmentField.get(null); + if (ciEnvMap != null) { + ciEnvMap.clear(); + ciEnvMap.putAll(env); + } + } } /** Prevents the CliFrontend from calling System.exit. */ From a61d4d6d099b1695629b6347f02eacbe98675a13 Mon Sep 17 00:00:00 2001 From: aIbrahiim Date: Fri, 10 Jul 2026 13:27:20 +0300 Subject: [PATCH 2/4] fix UnboundedSourceWrapperTest --- .../trigger_files/beam_PreCommit_Java.json | 2 +- .../io/UnboundedSourceWrapperTest.java | 20 ++++++++++++++----- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/.github/trigger_files/beam_PreCommit_Java.json b/.github/trigger_files/beam_PreCommit_Java.json index e797011e0364..a8f46c548f1e 100644 --- a/.github/trigger_files/beam_PreCommit_Java.json +++ b/.github/trigger_files/beam_PreCommit_Java.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 7 + "modification": 8 } diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java index f57198e08e3e..8a6b51fd3ba1 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java @@ -665,15 +665,16 @@ private static void testSourceDoesNotShutdown(boolean shouldHaveReaders) throws // Wait to see if the wrapper shuts down immediately in case it doesn't have readers if (!shouldHaveReaders) { // The expected state is for finalizeSource to sleep instead of exiting - while (true) { - StackTraceElement[] callStack = thread.getStackTrace(); - if (callStack.length >= 2 - && "sleep".equals(callStack[0].getMethodName()) - && "finalizeSource".equals(callStack[1].getMethodName())) { + long deadlineNs = System.nanoTime() + 9_000_000_000L; + boolean reachedFinalizeSource = false; + while (System.nanoTime() < deadlineNs) { + if (isInFinalizeSource(thread.getStackTrace())) { + reachedFinalizeSource = true; break; } Thread.sleep(10); } + assertThat(reachedFinalizeSource, is(true)); } // Source should still be running even if there are no readers assertThat(sourceWrapper.isRunning(), is(true)); @@ -694,6 +695,15 @@ private static void testSourceDoesNotShutdown(boolean shouldHaveReaders) throws assertThat(thread.isAlive(), is(false)); } + private static boolean isInFinalizeSource(StackTraceElement[] callStack) { + for (StackTraceElement frame : callStack) { + if ("finalizeSource".equals(frame.getMethodName())) { + return true; + } + } + return false; + } + @Test public void testSequentialReadingFromBoundedSource() throws Exception { UnboundedReadFromBoundedSource.BoundedToUnboundedSourceAdapter source = From dc1199989ae3316419f54303ae07895ee6cfac94 Mon Sep 17 00:00:00 2001 From: aIbrahiim Date: Fri, 10 Jul 2026 17:34:21 +0300 Subject: [PATCH 3/4] resolve comments --- .../trigger_files/beam_PreCommit_Java.json | 2 +- .../runners/flink/FlinkSubmissionTest.java | 35 +++++++++---------- .../io/UnboundedSourceWrapperTest.java | 6 +--- 3 files changed, 19 insertions(+), 24 deletions(-) diff --git a/.github/trigger_files/beam_PreCommit_Java.json b/.github/trigger_files/beam_PreCommit_Java.json index a8f46c548f1e..0e9f1cacf9bc 100644 --- a/.github/trigger_files/beam_PreCommit_Java.json +++ b/.github/trigger_files/beam_PreCommit_Java.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 8 + "modification": 9 } diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java index 538db8472d4a..feefe537c728 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java @@ -232,23 +232,26 @@ private static void restoreEnvironment() throws Exception { * because Flink's CliFrontend requires a Flink configuration file for which the location can only * be set using the {@code ConfigConstants.ENV_FLINK_CONF_DIR} environment variable. * - *

On Unix JDKs, {@code ProcessEnvironment.theEnvironment} is a {@code Map}. - * {@code System.getenv(String)} looks up {@code Variable} keys, so inserting plain {@code String} - * keys (via {@code putAll}) is a no-op for lookups and leaves Flink unable to find {@code - * FLINK_CONF_DIR}. On Windows the map uses {@code String} keys (plus a case-insensitive map). + *

On Unix, {@code theEnvironment} uses {@code Variable}/{@code Value} keys; {@code putAll} with + * {@code String} keys does not work for {@code System.getenv(String)} lookups. */ @SuppressWarnings("unchecked") private static void modifyEnv(Map env) throws Exception { - Class processEnvironment = Class.forName("java.lang.ProcessEnvironment"); - - Field theEnvironmentField = processEnvironment.getDeclaredField("theEnvironment"); + Class processEnv = Class.forName("java.lang.ProcessEnvironment"); + Field theEnvironmentField = processEnv.getDeclaredField("theEnvironment"); theEnvironmentField.setAccessible(true); Map envMap = (Map) theEnvironmentField.get(null); envMap.clear(); + + Class variableClass = null; + Class valueClass = null; try { - // Unix: keys/values are ProcessEnvironment$Variable / $Value, not String. - Class variableClass = Class.forName("java.lang.ProcessEnvironment$Variable"); - Class valueClass = Class.forName("java.lang.ProcessEnvironment$Value"); + variableClass = Class.forName("java.lang.ProcessEnvironment$Variable"); + valueClass = Class.forName("java.lang.ProcessEnvironment$Value"); + } catch (ClassNotFoundException e) { + // Windows: theEnvironment uses String keys. + } + if (variableClass != null && valueClass != null) { Method valueOfVariable = variableClass.getDeclaredMethod("valueOf", String.class); Method valueOfValue = valueClass.getDeclaredMethod("valueOf", String.class); valueOfVariable.setAccessible(true); @@ -258,18 +261,14 @@ private static void modifyEnv(Map env) throws Exception { valueOfVariable.invoke(null, entry.getKey()), valueOfValue.invoke(null, entry.getValue())); } - } catch (ClassNotFoundException e) { - // Windows (and similar): theEnvironment uses String keys. + } else { envMap.putAll(env); } - // Windows: System.getenv(String) reads the case-insensitive map. if (System.getProperty("os.name", "").toLowerCase(Locale.ROOT).startsWith("windows")) { - Field theCaseInsensitiveEnvironmentField = - processEnvironment.getDeclaredField("theCaseInsensitiveEnvironment"); - theCaseInsensitiveEnvironmentField.setAccessible(true); - Map ciEnvMap = - (Map) theCaseInsensitiveEnvironmentField.get(null); + Field ciEnvField = processEnv.getDeclaredField("theCaseInsensitiveEnvironment"); + ciEnvField.setAccessible(true); + Map ciEnvMap = (Map) ciEnvField.get(null); if (ciEnvMap != null) { ciEnvMap.clear(); ciEnvMap.putAll(env); diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java index 8a6b51fd3ba1..0033243d9255 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java @@ -665,16 +665,12 @@ private static void testSourceDoesNotShutdown(boolean shouldHaveReaders) throws // Wait to see if the wrapper shuts down immediately in case it doesn't have readers if (!shouldHaveReaders) { // The expected state is for finalizeSource to sleep instead of exiting - long deadlineNs = System.nanoTime() + 9_000_000_000L; - boolean reachedFinalizeSource = false; - while (System.nanoTime() < deadlineNs) { + while (true) { if (isInFinalizeSource(thread.getStackTrace())) { - reachedFinalizeSource = true; break; } Thread.sleep(10); } - assertThat(reachedFinalizeSource, is(true)); } // Source should still be running even if there are no readers assertThat(sourceWrapper.isRunning(), is(true)); From 632e92a514ce479ce5969b41d6e75cd882684041 Mon Sep 17 00:00:00 2001 From: aIbrahiim Date: Fri, 10 Jul 2026 18:58:37 +0300 Subject: [PATCH 4/4] Apply Spotless --- .../org/apache/beam/runners/flink/FlinkSubmissionTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java index feefe537c728..8e5ef8a3445a 100644 --- a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java @@ -232,8 +232,8 @@ private static void restoreEnvironment() throws Exception { * because Flink's CliFrontend requires a Flink configuration file for which the location can only * be set using the {@code ConfigConstants.ENV_FLINK_CONF_DIR} environment variable. * - *

On Unix, {@code theEnvironment} uses {@code Variable}/{@code Value} keys; {@code putAll} with - * {@code String} keys does not work for {@code System.getenv(String)} lookups. + *

On Unix, {@code theEnvironment} uses {@code Variable}/{@code Value} keys; {@code putAll} + * with {@code String} keys does not work for {@code System.getenv(String)} lookups. */ @SuppressWarnings("unchecked") private static void modifyEnv(Map env) throws Exception {