[beam] branch master updated: Automatically retrieving JDK 11 containers

[pa...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 399786f  Automatically retrieving JDK 11 containers
     new 9b1e580  Merge pull request #8031 from pabloem/j11-automatch
399786f is described below

commit 399786f2ef4e050b6b13cb57b552201eedaf2684
Author: pabloem <pa...@apache.org>
AuthorDate: Mon Mar 11 15:11:07 2019 -0700

    Automatically retrieving JDK 11 containers
---
 .../java/org/apache/beam/runners/dataflow/DataflowRunner.java    | 9 +++++++--
 .../org/apache/beam/runners/dataflow/DataflowRunnerTest.java     | 9 +++++++++
 2 files changed, 16 insertions(+), 2 deletions(-)

diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index a9b5f41..06bc9bd 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -1797,14 +1797,19 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
   @VisibleForTesting
   static String getContainerImageForJob(DataflowPipelineOptions options) {
     String workerHarnessContainerImage = options.getWorkerHarnessContainerImage();
+
+    String javaVersionId =
+        Float.parseFloat(System.getProperty("java.specification.version")) >= 9 ? "java11" : "java";
     if (!workerHarnessContainerImage.contains("IMAGE")) {
       return workerHarnessContainerImage;
     } else if (hasExperiment(options, "beam_fn_api")) {
       return workerHarnessContainerImage.replace("IMAGE", "java");
     } else if (options.isStreaming()) {
-      return workerHarnessContainerImage.replace("IMAGE", "beam-java-streaming");
+      return workerHarnessContainerImage.replace(
+          "IMAGE", String.format("beam-%s-streaming", javaVersionId));
     } else {
-      return workerHarnessContainerImage.replace("IMAGE", "beam-java-batch");
+      return workerHarnessContainerImage.replace(
+          "IMAGE", String.format("beam-%s-batch", javaVersionId));
     }
   }
 
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
index 00004f3..101d8d5 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
@@ -1337,10 +1337,19 @@ public class DataflowRunnerTest implements Serializable {
     options.setWorkerHarnessContainerImage("gcr.io/IMAGE/foo");
     options.setExperiments(null);
     options.setStreaming(false);
+    System.setProperty("java.specification.version", "1.8");
     assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java-batch/foo"));
+    // batch, legacy, jdk11
+    options.setStreaming(false);
+    System.setProperty("java.specification.version", "11");
+    assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java11-batch/foo"));
     // streaming, legacy
+    System.setProperty("java.specification.version", "1.8");
     options.setStreaming(true);
     assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java-streaming/foo"));
+    // streaming, legacy, jdk11
+    System.setProperty("java.specification.version", "11");
+    assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java11-streaming/foo"));
     // streaming, fnapi
     options.setExperiments(ImmutableList.of("experiment1", "beam_fn_api"));
     assertThat(getContainerImageForJob(options), equalTo("gcr.io/java/foo"));