@VisibleForTesting
  static String getContainerImageForJob(DataflowPipelineOptions options) {
    String containerImage = options.getSdkContainerImage();

    if (containerImage == null) {
      // If not set, construct and return default image URL.
      return getDefaultContainerImageUrl(options);
    } else if (containerImage.contains("IMAGE")) {
      // Replace placeholder with default image name
      return containerImage.replace("IMAGE", getDefaultContainerImageNameForJob(options));
    } else {
      return containerImage;
    }
  }
@Test
  public void testGetContainerImageForJobFromOptionWithPlaceholder() {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
    options.setSdkContainerImage("gcr.io/IMAGE/foo");

    for (Environments.JavaVersion javaVersion : Environments.JavaVersion.values()) {
      System.setProperty("java.specification.version", javaVersion.specification());
      // batch legacy
      options.setExperiments(null);
      options.setStreaming(false);
      assertThat(
          getContainerImageForJob(options),
          equalTo(String.format("gcr.io/beam-%s-batch/foo", javaVersion.legacyName())));

      // streaming, legacy
      options.setExperiments(null);
      options.setStreaming(true);
      assertThat(
          getContainerImageForJob(options),
          equalTo(String.format("gcr.io/beam-%s-streaming/foo", javaVersion.legacyName())));

      // batch, FnAPI
      options.setExperiments(ImmutableList.of("beam_fn_api"));
      options.setStreaming(false);
      assertThat(
          getContainerImageForJob(options),
          equalTo(String.format("gcr.io/beam_%s_sdk/foo", javaVersion.name())));

      // streaming, FnAPI
      options.setExperiments(ImmutableList.of("beam_fn_api"));
      options.setStreaming(true);
      assertThat(
          getContainerImageForJob(options),
          equalTo(String.format("gcr.io/beam_%s_sdk/foo", javaVersion.name())));
    }
  }