Add Spark JVM --add-opens for (Nexmark, TPC-DS, PortableJar)

Author: aIbrahiim

.github/trigger_files/beam_PostCommit_Java_Nexmark_Spark.json

@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "modification": 1
+}

.github/trigger_files/beam_PostCommit_Java_Tpcds_Spark.json

@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "modification": 1
+}

.github/trigger_files/beam_PostCommit_PortableJar_Spark.json

@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "modification": 1
+}

runners/portability/test_pipeline_jar.sh

@@ -123,7 +123,16 @@ OUTPUT_JAR="test-pipeline-${RUNNER}-$(date +%Y%m%d-%H%M%S).jar"
 
 if [[ "$TEST_EXIT_CODE" -eq 0 ]]; then
   # Execute the jar
-  java -jar $OUTPUT_JAR || TEST_EXIT_CODE=$?
+  JAVA_ARGS=()
+  if [[ "$RUNNER" = "SparkRunner" ]]; then
+    JAVA_ARGS+=(
+      "--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"
+    )
+  fi
+  java "${JAVA_ARGS[@]}" -jar $OUTPUT_JAR || TEST_EXIT_CODE=$?
 fi
 
 rm -rf $ENV_DIR

sdks/java/testing/nexmark/build.gradle

@@ -114,6 +114,19 @@ if (isSparkRunner) {
   }
 }
 
+def sparkJvmArgs() {
+  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"
+    ]
+  }
+  return []
+}
+
 def getNexmarkArgs = {
   def nexmarkArgsStr =  project.findProperty(nexmarkArgsProperty) ?: ""
   def nexmarkArgsList = new ArrayList<String>()
@@ -179,6 +192,7 @@ task run(type: JavaExec) {
     systemProperty "spark.ui.showConsoleProgress", "false"
     // Dataset runner only
     systemProperty "spark.sql.shuffle.partitions", "4"
+    jvmArgs += sparkJvmArgs()
   }
 
   mainClass = "org.apache.beam.sdk.nexmark.Main"

sdks/java/testing/tpcds/build.gradle

@@ -104,6 +104,19 @@ if (isSpark) {
     }
 }
 
+def sparkJvmArgs() {
+  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"
+    ]
+  }
+  return []
+}
+
 // Execute the TPC-DS queries or suites via Gradle.
 //
 // Parameters:
@@ -141,6 +154,7 @@ task run(type: JavaExec) {
     // Dataset runner only
     systemProperty "spark.sql.shuffle.partitions", "4"
     systemProperty "spark.sql.adaptive.enabled", "false" // high overhead for complex queries
+    jvmArgs += sparkJvmArgs()
   }
 
     mainClass = "org.apache.beam.sdk.tpcds.BeamTpcds"