Fix XVR JavaUsingPython tests using dev Beam at expansion (#36444)

* Uses test pipeline

* Set custom beam requirement

* Fix spotless

Author: Abacn

runners/flink/job-server/flink_job_server.gradle

@@ -269,6 +269,7 @@ createCrossLanguageValidatesRunnerTask(
     "--environmentCacheMillis=10000",
     "--experiments=beam_fn_api",
     "--parallelism=2",
+    "--customBeamRequirement=${project.project(":sdks:python").projectDir}/build/apache-beam.tar.gz",
   ],
   goScriptOptions: [
     "--runner flink",

runners/google-cloud-dataflow-java/build.gradle

@@ -486,6 +486,7 @@ createCrossLanguageValidatesRunnerTask(
     "--tempRoot=${dataflowValidatesTempRoot}",
     "--sdkContainerImage=${dockerJavaImageContainer}:${dockerTag}",
     "--sdkHarnessContainerImageOverrides=.*python.*,${dockerPythonImageContainer}:${dockerTag}",
+    "--customBeamRequirement=${project.project(":sdks:python").projectDir}/build/apache-beam.tar.gz",
   ],
   pytestOptions: [
     "--capture=no",

runners/samza/job-server/build.gradle

@@ -243,6 +243,7 @@ createCrossLanguageValidatesRunnerTask(
                 "--jobEndpoint=localhost:${jobPort}",
                 "--environmentCacheMillis=10000",
                 "--experiments=beam_fn_api",
+                "--customBeamRequirement=${project.project(":sdks:python").projectDir}/build/apache-beam.tar.gz",
         ],
         goScriptOptions: [
             "--runner samza",

runners/spark/job-server/spark_job_server.gradle

@@ -294,6 +294,7 @@ createCrossLanguageValidatesRunnerTask(
     "--jobEndpoint=localhost:${jobPort}",
     "--environmentCacheMillis=10000",
     "--experiments=beam_fn_api",
+    "--customBeamRequirement=${project.project(":sdks:python").projectDir}/build/apache-beam.tar.gz",
   ],
   goScriptOptions: [
     "--runner spark",

sdks/java/extensions/python/src/test/java/org/apache/beam/sdk/extensions/python/PythonExternalTransformTest.java

@@ -33,6 +33,7 @@
 import org.apache.beam.sdk.schemas.SchemaTranslation;
 import org.apache.beam.sdk.schemas.logicaltypes.MicrosInstant;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.testing.UsesPythonExpansionService;
 import org.apache.beam.sdk.testing.ValidatesRunner;
 import org.apache.beam.sdk.transforms.Create;
@@ -43,43 +44,45 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 @RunWith(JUnit4.class)
 public class PythonExternalTransformTest implements Serializable {
+  @Rule public transient TestPipeline testPipeline = TestPipeline.create();
 
   @Test
   @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
   public void trivialPythonTransform() {
-    Pipeline p = Pipeline.create();
     PCollection<String> output =
-        p.apply(Create.of(KV.of("A", "x"), KV.of("A", "y"), KV.of("B", "z")))
+        testPipeline
+            .apply(Create.of(KV.of("A", "x"), KV.of("A", "y"), KV.of("B", "z")))
             .apply(
                 PythonExternalTransform
                     .<PCollection<KV<String, String>>, PCollection<KV<String, Iterable<String>>>>
                         from("apache_beam.GroupByKey"))
             .apply(Keys.create());
     PAssert.that(output).containsInAnyOrder("A", "B");
-    // TODO: Run this on a multi-language supporting runner.
+    testPipeline.run();
   }
 
   @Test
   @Category({ValidatesRunner.class, UsesPythonExpansionService.class})
   public void pythonTransformWithDependencies() {
-    Pipeline p = Pipeline.create();
     PCollection<String> output =
-        p.apply(Create.of("elephant", "mouse", "sheep"))
+        testPipeline
+            .apply(Create.of("elephant", "mouse", "sheep"))
             .apply(
                 PythonExternalTransform.<PCollection<String>, PCollection<String>>from(
                         "apache_beam.Map")
                     .withArgs(PythonCallableSource.of("import inflection\ninflection.pluralize"))
                     .withExtraPackages(ImmutableList.of("inflection"))
                     .withOutputCoder(StringUtf8Coder.of()));
     PAssert.that(output).containsInAnyOrder("elephants", "mice", "sheep");
-    // TODO: Run this on a multi-language supporting runner.
+    testPipeline.run();
   }
 
   @Test

sdks/python/test-suites/direct/xlang/build.gradle

@@ -62,6 +62,7 @@ createCrossLanguageValidatesRunnerTask(
     "--jobEndpoint=localhost:${jobPort}",
     "--environmentCacheMillis=10000",
     "--experiments=beam_fn_api",
+    "--customBeamRequirement=${project.project(":sdks:python").projectDir}/build/apache-beam.tar.gz",
   ],
   goScriptOptions: [
     "--runner portable",