From 7d682e210f99133633c72ad12cd75f18b80ba7c1 Mon Sep 17 00:00:00 2001 From: Tomasz Wojdat Date: Fri, 10 Jul 2026 19:12:21 +0200 Subject: [PATCH 1/7] Bump httplib2 upper bound to `<1.0.0` (#39280) This allows for httplib2 0.32.0 which includes a fix for CVE-2026-59939. Resolves #39279. --- sdks/python/setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/setup.py b/sdks/python/setup.py index e764f935bea6..471f15723e70 100644 --- a/sdks/python/setup.py +++ b/sdks/python/setup.py @@ -421,7 +421,7 @@ def get_portability_package_data(): 'fastavro>=0.23.6,<2', 'fasteners>=0.3,<1.0', 'grpcio>=1.33.1,<2,!=1.48.0,!=1.59.*,!=1.60.*,!=1.61.*,!=1.62.0,!=1.62.1,!=1.66.*,!=1.67.*,!=1.68.*,!=1.69.*,!=1.70.*', # pylint: disable=line-too-long - 'httplib2>=0.8,<0.32.0', + 'httplib2>=0.8,<1.0.0', 'jsonpickle>=3.0.4,<5.0.0', # numpy can have breaking changes in minor versions. # Use a strict upper bound. From 2c92949b6a0d85f704517e20df064903c2628da5 Mon Sep 17 00:00:00 2001 From: Abdelrahman Ibrahim Date: Fri, 10 Jul 2026 20:18:21 +0300 Subject: [PATCH 2/7] pass --add-opens to Spark PVR test JVM on Java 21 (#39278) --- .../beam_PostCommit_Java_PVR_Spark3_Streaming.json | 2 +- runners/spark/job-server/spark_job_server.gradle | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json index f1ba03a243ee..455144f02a35 100644 --- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 5 + "modification": 6 } diff --git a/runners/spark/job-server/spark_job_server.gradle b/runners/spark/job-server/spark_job_server.gradle index e72b59444de8..5240bb310d05 100644 --- a/runners/spark/job-server/spark_job_server.gradle +++ b/runners/spark/job-server/spark_job_server.gradle @@ -280,6 +280,13 @@ def sparkJobServerJvmArgs() { return [] } +// TestPortableRunner starts SparkJobServerDriver in-process in the test JVM. +['validatesPortableRunnerDocker', 'validatesPortableRunnerBatch', 'validatesPortableRunnerStreaming'].each { taskName -> + tasks.named(taskName) { + jvmArgs += sparkJobServerJvmArgs() + } +} + def setupTask = project.tasks.register("sparkJobServerSetup", Exec) { dependsOn shadowJar def pythonDir = project.project(":sdks:python").projectDir From 89cc292fc185bc2b517b58889dbe3dba2f889621 Mon Sep 17 00:00:00 2001 From: Abdelrahman Ibrahim Date: Fri, 10 Jul 2026 20:23:15 +0300 Subject: [PATCH 3/7] Fix Flink runner tests on Java 21 (#39272) * fix Flink runner tests on Java 21 * fix UnboundedSourceWrapperTest * resolve comments * Apply Spotless --- .../trigger_files/beam_PreCommit_Java.json | 2 +- runners/flink/flink_runner.gradle | 17 +++++++ .../runners/flink/FlinkSubmissionTest.java | 51 +++++++++++++++---- .../io/UnboundedSourceWrapperTest.java | 14 +++-- 4 files changed, 68 insertions(+), 16 deletions(-) diff --git a/.github/trigger_files/beam_PreCommit_Java.json b/.github/trigger_files/beam_PreCommit_Java.json index 5abe02fc09c7..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": 1 + "modification": 9 } 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..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 @@ -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,49 @@ 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, {@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 processEnv = Class.forName("java.lang.ProcessEnvironment"); - Field envField = processEnv.getDeclaredField("theUnmodifiableEnvironment"); + Field theEnvironmentField = processEnv.getDeclaredField("theEnvironment"); + theEnvironmentField.setAccessible(true); + Map envMap = (Map) theEnvironmentField.get(null); + envMap.clear(); - 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); + Class variableClass = null; + Class valueClass = null; + try { + 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); + valueOfValue.setAccessible(true); + for (Map.Entry entry : env.entrySet()) { + envMap.put( + valueOfVariable.invoke(null, entry.getKey()), + valueOfValue.invoke(null, entry.getValue())); + } + } else { + envMap.putAll(env); + } - modifiersField.setInt(envField, envField.getModifiers() & Modifier.FINAL); - modifiersField.setAccessible(false); + if (System.getProperty("os.name", "").toLowerCase(Locale.ROOT).startsWith("windows")) { + Field ciEnvField = processEnv.getDeclaredField("theCaseInsensitiveEnvironment"); + ciEnvField.setAccessible(true); + Map ciEnvMap = (Map) ciEnvField.get(null); + if (ciEnvMap != null) { + ciEnvMap.clear(); + ciEnvMap.putAll(env); + } + } } /** Prevents the CliFrontend from calling System.exit. */ 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..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 @@ -666,10 +666,7 @@ private static void testSourceDoesNotShutdown(boolean shouldHaveReaders) throws 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())) { + if (isInFinalizeSource(thread.getStackTrace())) { break; } Thread.sleep(10); @@ -694,6 +691,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 63aecf2dc6744931011f3cda134d3f84eae43c4a Mon Sep 17 00:00:00 2001 From: tvalentyn Date: Fri, 10 Jul 2026 11:44:02 -0700 Subject: [PATCH 4/7] Add JVM arguments to fix hadoop-format tests on JDK 17+ (#39283) --- sdks/java/io/hadoop-format/build.gradle | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/sdks/java/io/hadoop-format/build.gradle b/sdks/java/io/hadoop-format/build.gradle index 73fc44a0f311..83dfcf22d86d 100644 --- a/sdks/java/io/hadoop-format/build.gradle +++ b/sdks/java/io/hadoop-format/build.gradle @@ -202,3 +202,24 @@ static def createTaskNames(Map hadoopVersions, String suffix) { .map{num -> "runHadoopFormatIO$num$suffix"} .collect(Collectors.toList()) } + +def testJavaVer = project.findProperty('testJavaVersion') ?: JavaVersion.current().getMajorVersion() +def isJava17OrHigher = testJavaVer.toInteger() >= 17 + +tasks.withType(Test).configureEach { + if (isJava17OrHigher) { + systemProperty 'java.security.manager', 'allow' + } + // Open and export modules for embedded Cassandra and Elasticsearch under Java 17+ + jvmArgs '--add-exports=java.base/sun.nio.ch=ALL-UNNAMED', + '--add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED', + '--add-exports=java.base/jdk.internal.misc=ALL-UNNAMED', + '--add-opens=java.base/sun.nio.ch=ALL-UNNAMED', + '--add-opens=java.base/java.lang=ALL-UNNAMED', + '--add-opens=java.base/java.util=ALL-UNNAMED', + '--add-opens=java.base/java.util.concurrent=ALL-UNNAMED', + '--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED', + '--add-opens=java.base/java.io=ALL-UNNAMED', + '--add-opens=java.base/java.net=ALL-UNNAMED', + '--add-opens=java.base/java.nio=ALL-UNNAMED' +} From 1081d592dcbcd73206453cda74262f694ce6d278 Mon Sep 17 00:00:00 2001 From: Ivy Xu Date: Sat, 11 Jul 2026 02:48:06 +0800 Subject: [PATCH 5/7] Remove previous workarounds for when `super()` could not be pickled on Py3 (#39187) * Remove the workaround for pickle * Remove the workaround for pickle --- .../examples/complete/autocomplete.py | 5 ++-- .../examples/complete/game/game_stats.py | 14 ++++------- .../complete/game/hourly_team_score.py | 19 ++++++--------- .../examples/complete/game/leader_board.py | 24 +++++++------------ .../examples/complete/game/user_score.py | 9 +++---- .../complete/top_wikipedia_sessions.py | 5 ++-- .../examples/cookbook/bigtableio_it_test.py | 4 +--- .../examples/wordcount_debugging.py | 4 +--- .../examples/wordcount_with_metrics.py | 4 +--- 9 files changed, 31 insertions(+), 57 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/autocomplete.py b/sdks/python/apache_beam/examples/complete/autocomplete.py index 4e4c5143b96b..5db2617d6794 100644 --- a/sdks/python/apache_beam/examples/complete/autocomplete.py +++ b/sdks/python/apache_beam/examples/complete/autocomplete.py @@ -58,9 +58,8 @@ def format_result(prefix_candidates): class TopPerPrefix(beam.PTransform): def __init__(self, count): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self._count = count def expand(self, words): diff --git a/sdks/python/apache_beam/examples/complete/game/game_stats.py b/sdks/python/apache_beam/examples/complete/game/game_stats.py index 233d22b75427..ad63fba1be01 100644 --- a/sdks/python/apache_beam/examples/complete/game/game_stats.py +++ b/sdks/python/apache_beam/examples/complete/game/game_stats.py @@ -105,9 +105,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): @@ -131,9 +129,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): @@ -171,9 +168,8 @@ def __init__(self, table_name, dataset, schema, project): schema: Dictionary in the format {'column_name': 'bigquery_type'} project: Name of the Cloud project containing BigQuery table. """ - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.table_name = table_name self.dataset = dataset self.schema = schema diff --git a/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py b/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py index 48a105af527d..8542350f0cd5 100644 --- a/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py +++ b/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py @@ -105,9 +105,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): @@ -131,9 +129,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): @@ -171,9 +168,8 @@ def __init__(self, table_name, dataset, schema, project): schema: Dictionary in the format {'column_name': 'bigquery_type'} project: Name of the Cloud project containing BigQuery table. """ - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.table_name = table_name self.dataset = dataset self.schema = schema @@ -196,9 +192,8 @@ def expand(self, pcoll): # [START main] class HourlyTeamScore(beam.PTransform): def __init__(self, start_min, stop_min, window_duration): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.start_timestamp = str2timestamp(start_min) self.stop_timestamp = str2timestamp(stop_min) self.window_duration_in_seconds = window_duration * 60 diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board.py b/sdks/python/apache_beam/examples/complete/game/leader_board.py index 308e1e1cf5c0..96cde175c409 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board.py @@ -114,9 +114,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): @@ -140,9 +138,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): @@ -180,9 +177,8 @@ def __init__(self, table_name, dataset, schema, project): schema: Dictionary in the format {'column_name': 'bigquery_type'} project: Name of the Cloud project containing BigQuery table. """ - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.table_name = table_name self.dataset = dataset self.schema = schema @@ -210,9 +206,8 @@ class CalculateTeamScores(beam.PTransform): default. """ def __init__(self, team_window_duration, allowed_lateness): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.team_window_duration = team_window_duration * 60 self.allowed_lateness_seconds = allowed_lateness * 60 @@ -242,9 +237,8 @@ class CalculateUserScores(beam.PTransform): global windowing. Get periodic updates on all users' running scores. """ def __init__(self, allowed_lateness): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.allowed_lateness_seconds = allowed_lateness * 60 def expand(self, pcoll): diff --git a/sdks/python/apache_beam/examples/complete/game/user_score.py b/sdks/python/apache_beam/examples/complete/game/user_score.py index 03f0d00fc30f..ae021b45f560 100644 --- a/sdks/python/apache_beam/examples/complete/game/user_score.py +++ b/sdks/python/apache_beam/examples/complete/game/user_score.py @@ -97,9 +97,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): @@ -124,9 +122,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): diff --git a/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py b/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py index 50b026edf240..440a13fa9a0e 100644 --- a/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py +++ b/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py @@ -111,9 +111,8 @@ def format_output(element, window=beam.DoFn.WindowParam): class ComputeTopSessions(beam.PTransform): """Computes the top user sessions for each month.""" def __init__(self, sampling_threshold): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.sampling_threshold = sampling_threshold def expand(self, pcoll): diff --git a/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py b/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py index cc11ec071cc1..84087dee5607 100644 --- a/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py +++ b/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py @@ -66,9 +66,7 @@ class GenerateTestRows(beam.PTransform): """ def __init__(self, number, project_id=None, instance_id=None, table_id=None): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() self.number = number self.rand = random.choice(string.ascii_letters + string.digits) self.column_family_id = 'cf1' diff --git a/sdks/python/apache_beam/examples/wordcount_debugging.py b/sdks/python/apache_beam/examples/wordcount_debugging.py index 581bbd3adc1b..9fb446a90a1b 100644 --- a/sdks/python/apache_beam/examples/wordcount_debugging.py +++ b/sdks/python/apache_beam/examples/wordcount_debugging.py @@ -79,9 +79,7 @@ class FilterTextFn(beam.DoFn): """A DoFn that filters for a specific key based on a regular expression.""" def __init__(self, pattern): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.pattern = pattern # A custom metric can track values in your pipeline as it runs. Those # values will be available in the monitoring system of the runner used diff --git a/sdks/python/apache_beam/examples/wordcount_with_metrics.py b/sdks/python/apache_beam/examples/wordcount_with_metrics.py index f575a8d7fbba..e26a627488eb 100644 --- a/sdks/python/apache_beam/examples/wordcount_with_metrics.py +++ b/sdks/python/apache_beam/examples/wordcount_with_metrics.py @@ -53,9 +53,7 @@ class WordExtractingDoFn(beam.DoFn): """Parse each line of input text into words.""" def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.words_counter = Metrics.counter(self.__class__, 'words') self.word_lengths_counter = Metrics.counter(self.__class__, 'word_lengths') self.word_lengths_dist = Metrics.distribution( From a3de8e1aa30777848596cf742a957db246cf8e97 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Sat, 11 Jul 2026 00:28:13 +0400 Subject: [PATCH 6/7] Remove redundant argument (#39289) --- .github/workflows/beam_PostCommit_Java_Jpms_Dataflow.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/workflows/beam_PostCommit_Java_Jpms_Dataflow.yml b/.github/workflows/beam_PostCommit_Java_Jpms_Dataflow.yml index 17f7cd2d5881..0fe615bb2d0d 100644 --- a/.github/workflows/beam_PostCommit_Java_Jpms_Dataflow.yml +++ b/.github/workflows/beam_PostCommit_Java_Jpms_Dataflow.yml @@ -76,7 +76,6 @@ jobs: uses: ./.github/actions/gradle-command-self-hosted-action with: gradle-command: :sdks:java:testing:jpms-tests:dataflowRunnerIntegrationTest - arguments: -Dorg.gradle.java.home=$JAVA_HOME_11_X64 - name: Archive JUnit Test Results uses: actions/upload-artifact@v7 if: ${{ !success() }} From cccb64946675deeb2e77f9f3d5ced2ab63e4f6e2 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Sat, 11 Jul 2026 00:28:58 +0400 Subject: [PATCH 7/7] Fix Community Metrics Prober job (#39290) --- .test-infra/metrics/src/test/groovy/ProberTests.groovy | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.test-infra/metrics/src/test/groovy/ProberTests.groovy b/.test-infra/metrics/src/test/groovy/ProberTests.groovy index c5de9ca64c8a..6f9befc2c201 100644 --- a/.test-infra/metrics/src/test/groovy/ProberTests.groovy +++ b/.test-infra/metrics/src/test/groovy/ProberTests.groovy @@ -22,12 +22,12 @@ import groovy.json.JsonSlurper import static groovy.test.GroovyAssert.shouldFail /** - * Prober tests which performs health checks on deployed infrasture for + * Prober tests which performs health checks on deployed infrastructure for * community metrics. */ class ProberTests { // TODO: Make this configurable - def grafanaEndpoint = 'http://metrics.beam.apache.org' + def grafanaEndpoint = 'https://metrics.beam.apache.org' @Test void PingGrafanaHttpApi() { @@ -36,14 +36,14 @@ class ProberTests { def dashboardNames = allDashboards.title // Validate at least one expected dashboard exists assert dashboardNames.contains('Post-commit Test Reliability') : 'Expected dashboard does not exist' - assert dashboardNames.size > 0 : "No dashboards found. Check Grafana dashboard initialization script." + assert dashboardNames.size() > 0 : "No dashboards found. Check Grafana dashboard initialization script." } @Test void CheckGrafanaStalenessAlerts() { def alertsJson = "${grafanaEndpoint}/api/alerts?dashboardQuery=Source%20Data%20Freshness".toURL().text def alerts = new JsonSlurper().parseText(alertsJson) - assert alerts.size > 0 + assert alerts.size() > 0 alerts.each { alert -> assert alert.state == 'ok' : "Input data is stale! ${alert}\n See: ${grafanaEndpoint}/d/data-freshness" }