Add idle_shutdown_timeout=5m flag

apache · Aug 23, 2024 · 40813b3 · 40813b3
1 parent 02c94dd
commit 40813b3
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 1 deletion.
diff --git a/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismExecutor.java b/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismExecutor.java
@@ -45,6 +45,7 @@
 abstract class PrismExecutor {
 
   private static final Logger LOG = LoggerFactory.getLogger(PrismExecutor.class);
+  static final String IDLE_SHUTDOWN_TIMEOUT = "-idle_shutdown_timeout=%s";
   static final String JOB_PORT_FLAG_TEMPLATE = "-job_port=%s";
   static final String SERVE_HTTP_FLAG_TEMPLATE = "-serve_http=%s";
 

diff --git a/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismPipelineOptions.java b/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismPipelineOptions.java
@@ -51,4 +51,11 @@ public interface PrismPipelineOptions extends PortablePipelineOptions {
   Boolean getEnableWebUI();
 
   void setEnableWebUI(Boolean enableWebUI);
+
+  @Description(
+      "Duration, represented as a String, that prism will wait for a new job before shutting itself down. Negative durations disable auto shutdown. Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\".")
+  @Default.String("5m")
+  String getIdleShutdownTimeout();
+
+  void setIdleShutdownTimeout(String idleShutdownTimeout);
 }
diff --git a/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismRunner.java b/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismRunner.java
@@ -98,13 +98,16 @@ PrismExecutor startPrism() throws IOException {
     String serveHttpFlag =
         String.format(
             PrismExecutor.SERVE_HTTP_FLAG_TEMPLATE, prismPipelineOptions.getEnableWebUI());
+    String idleShutdownTimeoutFlag =
+        String.format(
+            PrismExecutor.IDLE_SHUTDOWN_TIMEOUT, prismPipelineOptions.getIdleShutdownTimeout());
     String endpoint = "localhost:" + port;
     prismPipelineOptions.setJobEndpoint(endpoint);
     String command = locator.resolve();
     PrismExecutor executor =
         PrismExecutor.builder()
             .setCommand(command)
-            .setArguments(Arrays.asList(portFlag, serveHttpFlag))
+            .setArguments(Arrays.asList(portFlag, serveHttpFlag, idleShutdownTimeoutFlag))
             .build();
     executor.execute();
     checkState(executor.isAlive());