@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 testGetContainerImageForJobFromOption() {
    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

    String[] testCases = {
      "some-container",

      // It is important that empty string is preserved, as
      // dataflowWorkerJar relies on being passed an empty value vs
      // not providing the container image option at all.
      "",
    };

    for (String testCase : testCases) {
      // When image option is set, should use that exact image.
      options.setSdkContainerImage(testCase);
      assertThat(getContainerImageForJob(options), equalTo(testCase));
    }
  }