Finalize Java Prism Runner

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -628,6 +628,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def nemo_version = "0.1"
     // [bomupgrader] determined by: io.grpc:grpc-netty, consistent with: google_cloud_platform_libraries_bom
     def netty_version = "4.1.100.Final"
+    def nuprocess_version = "2.0.6"
     def postgres_version = "42.2.16"
     def powermock_version = "2.0.9"
     // [bomupgrader] determined by: com.google.protobuf:protobuf-java, consistent with: google_cloud_platform_libraries_bom
@@ -848,6 +849,7 @@ class BeamModulePlugin implements Plugin<Project> {
         netty_tcnative_boringssl_static             : "io.netty:netty-tcnative-boringssl-static:2.0.52.Final",
         netty_transport                             : "io.netty:netty-transport:$netty_version",
         netty_transport_native_epoll                : "io.netty:netty-transport-native-epoll:$netty_version",
+        nuprocess                                   : "com.zaxxer:nuprocess:$nuprocess_version",
         postgres                                    : "org.postgresql:postgresql:$postgres_version",
         powermock                                   : "org.powermock:powermock-module-junit4:$powermock_version",
         powermock_mockito                           : "org.powermock:powermock-api-mockito2:$powermock_version",

runners/portability/java/src/main/java/org/apache/beam/runners/portability/JobServicePipelineResult.java

@@ -40,7 +40,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class JobServicePipelineResult implements PipelineResult, AutoCloseable {
+public class JobServicePipelineResult implements PipelineResult, AutoCloseable {
 
   private static final long POLL_INTERVAL_MS = 3_000;
 
@@ -116,7 +116,7 @@ void setTerminalStateFuture(CompletableFuture<State> terminalStateFuture) {
     this.terminalStateFuture = terminalStateFuture;
   }
 
-  CompletableFuture<State> getTerminalStateFuture() {
+  public CompletableFuture<State> getTerminalStateFuture() {
     return this.terminalStateFuture;
   }
 
@@ -193,6 +193,14 @@ public MetricResults metrics() {
 
   @Override
   public void close() {
+    LOG.info("starting cleanup");
     cleanup.run();
+    LOG.info("finished cleanup");
+    executorService.shutdown();
+    try {
+      boolean ignored = executorService.awaitTermination(3000L, TimeUnit.MILLISECONDS);
+    } catch (InterruptedException ignore) {
+    }
+    LOG.info("shutdown executor service");
   }
 }

runners/prism/java/build.gradle

@@ -18,7 +18,7 @@
 
 plugins { id 'org.apache.beam.module' }
 
-applyJavaNature(
+applyPortabilityNature (
         automaticModuleName: 'org.apache.beam.runners.prism',
 )
 
@@ -37,6 +37,7 @@ dependencies {
     implementation library.java.slf4j_api
     implementation library.java.vendored_grpc_1_60_1
     implementation library.java.vendored_guava_32_1_2_jre
+    implementation library.java.nuprocess
 
     testImplementation library.java.junit
     testImplementation library.java.mockito_core