[pull] master from apache:master

.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
 }

.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
 }

.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() }}

.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"
     }

runners/flink/flink_runner.gradle

@@ -180,13 +180,30 @@ 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:
   // systemProperty "org.slf4j.simpleLogger.defaultLogLevel", "debug"
   // 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")
   }

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.
+   *
+   * <p>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<String, String> env) throws Exception {
     Class processEnv = Class.forName("java.lang.ProcessEnvironment");
-    Field envField = processEnv.getDeclaredField("theUnmodifiableEnvironment");
+    Field theEnvironmentField = processEnv.getDeclaredField("theEnvironment");
+    theEnvironmentField.setAccessible(true);
+    Map<Object, Object> envMap = (Map<Object, Object>) 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<String, String> 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<String, String> ciEnvMap = (Map<String, String>) ciEnvField.get(null);
+      if (ciEnvMap != null) {
+        ciEnvMap.clear();
+        ciEnvMap.putAll(env);
+      }
+    }
   }
 
   /** Prevents the CliFrontend from calling System.exit. */

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<Long> source =

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

sdks/java/io/hadoop-format/build.gradle

@@ -202,3 +202,24 @@ static def createTaskNames(Map<String, String> 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'
+}

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):

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

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

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):

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):