(beam) branch addjava21 updated: Honor JAVA_HOME in JavaJarServer checks and tests

[yh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yhu pushed a commit to branch addjava21
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/addjava21 by this push:
     new 5abaafe1106 Honor JAVA_HOME in JavaJarServer checks and tests
5abaafe1106 is described below

commit 5abaafe110621250e99aa089e50eeefcb8e84185
Author: Yi Hu <ya...@google.com>
AuthorDate: Wed Jun 11 17:06:22 2025 -0400

    Honor JAVA_HOME in JavaJarServer checks and tests
---
 .github/workflows/beam_PostCommit_Python.yml             |  5 ++++-
 runners/flink/job-server/flink_job_server.gradle         |  6 ++++++
 .../apache_beam/io/external/xlang_kafkaio_it_test.py     | 10 ++++++++--
 sdks/python/apache_beam/io/kafka.py                      |  3 ++-
 sdks/python/apache_beam/io/kinesis.py                    |  3 ++-
 sdks/python/apache_beam/io/snowflake.py                  |  3 ++-
 .../apache_beam/runners/portability/samza_runner_test.py |  3 ++-
 .../apache_beam/runners/portability/spark_runner_test.py |  3 ++-
 sdks/python/apache_beam/transforms/external.py           |  1 +
 sdks/python/apache_beam/transforms/external_java.py      |  4 +++-
 sdks/python/apache_beam/utils/subprocess_server.py       | 16 +++++++++++-----
 sdks/python/apache_beam/yaml/yaml_provider.py            |  3 ++-
 sdks/python/test-suites/portable/common.gradle           | 13 +++++++++++++
 13 files changed, 58 insertions(+), 15 deletions(-)

diff --git a/.github/workflows/beam_PostCommit_Python.yml b/.github/workflows/beam_PostCommit_Python.yml
index af34428aea0..2a98ccb0efb 100644
--- a/.github/workflows/beam_PostCommit_Python.yml
+++ b/.github/workflows/beam_PostCommit_Python.yml
@@ -78,7 +78,9 @@ jobs:
       - name: Setup environment
         uses: ./.github/actions/setup-environment-action
         with:
-          java-version: 21
+          java-version: |
+            21
+            11
           python-version: ${{ matrix.python_version }}
       - name: Install docker compose
         run: |
@@ -95,6 +97,7 @@ jobs:
         with:
           gradle-command: :python${{steps.set_py_ver_clean.outputs.py_ver_clean}}PostCommit
           arguments: |
+            -Pjava21Home=$JAVA_HOME_21_X64 \
             -PuseWheelDistribution \
             -PpythonVersion=${{ matrix.python_version }} \
         env:
diff --git a/runners/flink/job-server/flink_job_server.gradle b/runners/flink/job-server/flink_job_server.gradle
index 1c610477a44..90890a7d585 100644
--- a/runners/flink/job-server/flink_job_server.gradle
+++ b/runners/flink/job-server/flink_job_server.gradle
@@ -277,6 +277,12 @@ createCrossLanguageValidatesRunnerTask(
   ],
 )
 
+shadowJar {
+    manifest {
+        attributes(["Multi-Release": true])
+    }
+}
+
 // miniCluster jar starts an embedded Flink cluster intended for use in testing.
 tasks.register("miniCluster", Jar) {
   dependsOn shadowJar
diff --git a/sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py b/sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
index a7bf686d064..23178b0ee36 100644
--- a/sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
+++ b/sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
@@ -40,6 +40,7 @@ from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
 from apache_beam.transforms.userstate import BagStateSpec
 from apache_beam.transforms.userstate import CombiningValueStateSpec
+from apache_beam.utils import subprocess_server
 
 NUM_RECORDS = 1000
 
@@ -220,8 +221,13 @@ class CrossLanguageKafkaIOTest(unittest.TestCase):
     zookeeper_port = str(self.get_open_port())
     kafka_server = None
     try:
-      kafka_server = subprocess.Popen(
-          ['java', '-jar', local_kafka_jar_file, kafka_port, zookeeper_port])
+      kafka_server = subprocess.Popen([
+          subprocess_server.JavaHelper.get_java(),
+          '-jar',
+          local_kafka_jar_file,
+          kafka_port,
+          zookeeper_port
+      ])
       time.sleep(3)
       yield kafka_port
     finally:
diff --git a/sdks/python/apache_beam/io/kafka.py b/sdks/python/apache_beam/io/kafka.py
index 3ed4737cd76..d6be00f0502 100644
--- a/sdks/python/apache_beam/io/kafka.py
+++ b/sdks/python/apache_beam/io/kafka.py
@@ -45,7 +45,8 @@
   pipeline.
 
   * Install Java runtime in the computer from where the pipeline is constructed
-    and make sure that 'java' command is available.
+    and make sure that 'java' command is available or set JAVA_HOME environment
+    variable.
 
   In this option, Python SDK will either download (for released Beam version) or
   build (when running from a Beam Git clone) an expansion service jar and use
diff --git a/sdks/python/apache_beam/io/kinesis.py b/sdks/python/apache_beam/io/kinesis.py
index a066ec5cb2f..b4c939703f9 100644
--- a/sdks/python/apache_beam/io/kinesis.py
+++ b/sdks/python/apache_beam/io/kinesis.py
@@ -44,7 +44,8 @@
   pipeline.
 
   * Install Java runtime in the computer from where the pipeline is constructed
-    and make sure that 'java' command is available.
+    and make sure that 'java' command is available or set JAVA_HOME environment
+    variable.
 
   In this option, Python SDK will either download (for released Beam version) or
   build (when running from a Beam Git clone) a expansion service jar and use
diff --git a/sdks/python/apache_beam/io/snowflake.py b/sdks/python/apache_beam/io/snowflake.py
index d789a72001d..7f85af5ac69 100644
--- a/sdks/python/apache_beam/io/snowflake.py
+++ b/sdks/python/apache_beam/io/snowflake.py
@@ -39,7 +39,8 @@
   before running the Beam pipeline.
 
   * Install Java runtime in the computer from where the pipeline is constructed
-    and make sure that 'java' command is available.
+    and make sure that 'java' command is available or set JAVA_HOME environment
+    variable.
 
   In this option, Python SDK will either download (for released Beam version) or
   build (when running from a Beam Git clone) a expansion service jar and use
diff --git a/sdks/python/apache_beam/runners/portability/samza_runner_test.py b/sdks/python/apache_beam/runners/portability/samza_runner_test.py
index e46885c7c96..6112a50e851 100644
--- a/sdks/python/apache_beam/runners/portability/samza_runner_test.py
+++ b/sdks/python/apache_beam/runners/portability/samza_runner_test.py
@@ -33,6 +33,7 @@ from apache_beam.options.pipeline_options import PortableOptions
 from apache_beam.runners.portability import job_server
 from apache_beam.runners.portability import portable_runner
 from apache_beam.runners.portability import portable_runner_test
+from apache_beam.utils import subprocess_server
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -97,7 +98,7 @@ class SamzaRunnerTest(portable_runner_test.PortableRunnerTest):
 
     try:
       return [
-          'java',
+          subprocess_server.JavaHelper.get_java(),
           '-jar',
           cls.samza_job_server_jar,
           '--artifacts-dir',
diff --git a/sdks/python/apache_beam/runners/portability/spark_runner_test.py b/sdks/python/apache_beam/runners/portability/spark_runner_test.py
index 9fc7da2449f..cf6536be4c2 100644
--- a/sdks/python/apache_beam/runners/portability/spark_runner_test.py
+++ b/sdks/python/apache_beam/runners/portability/spark_runner_test.py
@@ -29,6 +29,7 @@ from apache_beam.options.pipeline_options import PortableOptions
 from apache_beam.runners.portability import job_server
 from apache_beam.runners.portability import portable_runner
 from apache_beam.runners.portability import portable_runner_test
+from apache_beam.utils import subprocess_server
 
 # Run as
 #
@@ -98,7 +99,7 @@ class SparkRunnerTest(portable_runner_test.PortableRunnerTest):
 
     try:
       return [
-          'java',
+          subprocess_server.JavaHelper.get_java(),
           '-Dbeam.spark.test.reuseSparkContext=true',
           '-jar',
           cls.spark_job_server_jar,
diff --git a/sdks/python/apache_beam/transforms/external.py b/sdks/python/apache_beam/transforms/external.py
index f6d361a9eca..c8f9b1d4515 100644
--- a/sdks/python/apache_beam/transforms/external.py
+++ b/sdks/python/apache_beam/transforms/external.py
@@ -1154,6 +1154,7 @@ def _maybe_use_transform_service(provided_service=None, options=None):
     return provided_service
 
   def is_java_available():
+    java = subprocess_server.JavaHelper.get_java()
     cmd = ['java', '--version']
 
     try:
diff --git a/sdks/python/apache_beam/transforms/external_java.py b/sdks/python/apache_beam/transforms/external_java.py
index e3984fa8ef2..ebd760f70f7 100644
--- a/sdks/python/apache_beam/transforms/external_java.py
+++ b/sdks/python/apache_beam/transforms/external_java.py
@@ -32,6 +32,7 @@ from apache_beam.testing.test_pipeline import TestPipeline
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
 from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder
+from apache_beam.utils import subprocess_server
 
 # Protect against environments where apitools library is not available.
 # pylint: disable=wrong-import-order, wrong-import-position
@@ -67,7 +68,8 @@ class JavaExternalTransformTest(object):
 
       # Start the java server and wait for it to be ready.
       if jar:
-        self._server = subprocess.Popen(['java', '-jar', jar, str(port)])
+        java = subprocess_server.JavaHelper.get_java()
+        self._server = subprocess.Popen([java, '-jar', jar, str(port)])
 
       address = 'localhost:%s' % str(port)
 
diff --git a/sdks/python/apache_beam/utils/subprocess_server.py b/sdks/python/apache_beam/utils/subprocess_server.py
index 85d9286bddd..babe81d6bde 100644
--- a/sdks/python/apache_beam/utils/subprocess_server.py
+++ b/sdks/python/apache_beam/utils/subprocess_server.py
@@ -113,6 +113,16 @@ class _SharedCache:
       return self._cache[key].obj
 
 
+class JavaHelper:
+  @classmethod
+  def get_java(cls):
+    java_path = 'java'
+    java_home = os.environ.get('JAVA_HOME')
+    if java_home:
+      java_path = os.path.join(java_home, 'bin', 'java')
+    return java_path
+
+
 class SubprocessServer(object):
   """An abstract base class for running GRPC Servers as an external process.
 
@@ -274,11 +284,7 @@ class JavaJarServer(SubprocessServer):
       java_arguments,
       classpath=None,
       cache_dir=None):
-    java_path = 'java'
-    java_home = os.environ.get('JAVA_HOME')
-    if java_home:
-      java_path = os.path.join(java_home, 'bin', 'java')
-    self._java_path = java_path
+    self._java_path = JavaHelper.get_java()
     if classpath:
       # java -jar ignores the classpath, so we make a new jar that embeds
       # the requested classpath.
diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py
index d7c427228a3..7c8114b5770 100755
--- a/sdks/python/apache_beam/yaml/yaml_provider.py
+++ b/sdks/python/apache_beam/yaml/yaml_provider.py
@@ -385,7 +385,8 @@ class ExternalJavaProvider(ExternalProvider):
 
   def available(self):
     # pylint: disable=subprocess-run-check
-    trial = subprocess.run(['which', 'java'], capture_output=True)
+    trial = subprocess.run(['which', subprocess_server.JavaHelper.get_java()],
+                           capture_output=True)
     if trial.returncode == 0:
       return True
     else:
diff --git a/sdks/python/test-suites/portable/common.gradle b/sdks/python/test-suites/portable/common.gradle
index 4211cd00947..a6bff7f4fbd 100644
--- a/sdks/python/test-suites/portable/common.gradle
+++ b/sdks/python/test-suites/portable/common.gradle
@@ -412,7 +412,20 @@ project.tasks.register("postCommitPy${pythonVersionSuffix}IT") {
             "pipeline_opts": pipelineOpts.join(" "),
     ])
     def kafkaJar = project(":sdks:java:testing:kafka-service:").buildTestKafkaServiceJar.archivePath
+
+    String fork_java_home = null
+    if (JavaVersion.current() < JavaVersion.VERSION_17) {
+      if (project.hasProperty("java17Home")) {
+        fork_java_home = project.getProperty("java17Home")
+      } else if (project.hasProperty("java21Home")) {
+        fork_java_home = project.getProperty("java21Home")
+      }
+    }
+
     exec {
+      if (fork_java_home != null) {
+        environment "JAVA_HOME", fork_java_home
+      }
       environment "LOCAL_KAFKA_JAR", kafkaJar
       executable 'sh'
       args '-c', ". ${envdir}/bin/activate && ${pythonRootDir}/scripts/run_integration_test.sh $cmdArgs"