protected RunnerApi.Pipeline applySdkEnvironmentOverrides(
      RunnerApi.Pipeline pipeline, DataflowPipelineOptions options) {
    String sdkHarnessContainerImageOverrides = options.getSdkHarnessContainerImageOverrides();
    String[] overrides =
        Strings.isNullOrEmpty(sdkHarnessContainerImageOverrides)
            ? new String[0]
            : sdkHarnessContainerImageOverrides.split(",", -1);
    if (overrides.length % 2 != 0) {
      throw new RuntimeException(
          "invalid syntax for SdkHarnessContainerImageOverrides: "
              + options.getSdkHarnessContainerImageOverrides());
    }
    RunnerApi.Pipeline.Builder pipelineBuilder = pipeline.toBuilder();
    RunnerApi.Components.Builder componentsBuilder = pipelineBuilder.getComponentsBuilder();
    componentsBuilder.clearEnvironments();
    for (Map.Entry<String, RunnerApi.Environment> entry :
        pipeline.getComponents().getEnvironmentsMap().entrySet()) {
      RunnerApi.Environment.Builder environmentBuilder = entry.getValue().toBuilder();
      if (BeamUrns.getUrn(RunnerApi.StandardEnvironments.Environments.DOCKER)
          .equals(environmentBuilder.getUrn())) {
        RunnerApi.DockerPayload dockerPayload;
        try {
          dockerPayload = RunnerApi.DockerPayload.parseFrom(environmentBuilder.getPayload());
        } catch (InvalidProtocolBufferException e) {
          throw new RuntimeException("Error parsing environment docker payload.", e);
        }
        String containerImage = dockerPayload.getContainerImage();
        boolean updated = false;
        for (int i = 0; i < overrides.length; i += 2) {
          containerImage = containerImage.replaceAll(overrides[i], overrides[i + 1]);
          if (!containerImage.equals(dockerPayload.getContainerImage())) {
            updated = true;
          }
        }
        if (containerImage.startsWith("apache/beam")
            && !updated
            // don't update if the container image is already configured by DataflowRunner
            && !containerImage.equals(getContainerImageForJob(options))) {
          containerImage =
              DataflowRunnerInfo.getDataflowRunnerInfo().getContainerImageBaseRepository()
                  + containerImage.substring(containerImage.lastIndexOf("/"));
        }
        environmentBuilder.setPayload(
            RunnerApi.DockerPayload.newBuilder()
                .setContainerImage(containerImage)
                .build()
                .toByteString());
      }
      componentsBuilder.putEnvironments(entry.getKey(), environmentBuilder.build());
    }
    return pipelineBuilder.build();
  }
@Test
  public void testApplySdkEnvironmentOverrides() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    String dockerHubPythonContainerUrl = "apache/beam_python3.8_sdk:latest";
    String gcrPythonContainerUrl = "gcr.io/apache-beam-testing/beam-sdk/beam_python3.8_sdk:latest";
    options.setSdkHarnessContainerImageOverrides(".*python.*," + gcrPythonContainerUrl);
    DataflowRunner runner = DataflowRunner.fromOptions(options);
    RunnerApi.Pipeline pipeline =
        RunnerApi.Pipeline.newBuilder()
            .setComponents(
                RunnerApi.Components.newBuilder()
                    .putEnvironments(
                        "env",
                        RunnerApi.Environment.newBuilder()
                            .setUrn(
                                BeamUrns.getUrn(RunnerApi.StandardEnvironments.Environments.DOCKER))
                            .setPayload(
                                RunnerApi.DockerPayload.newBuilder()
                                    .setContainerImage(dockerHubPythonContainerUrl)
                                    .build()
                                    .toByteString())
                            .build()))
            .build();
    RunnerApi.Pipeline expectedPipeline =
        RunnerApi.Pipeline.newBuilder()
            .setComponents(
                RunnerApi.Components.newBuilder()
                    .putEnvironments(
                        "env",
                        RunnerApi.Environment.newBuilder()
                            .setUrn(
                                BeamUrns.getUrn(RunnerApi.StandardEnvironments.Environments.DOCKER))
                            .setPayload(
                                RunnerApi.DockerPayload.newBuilder()
                                    .setContainerImage(gcrPythonContainerUrl)
                                    .build()
                                    .toByteString())
                            .build()))
            .build();
    assertThat(runner.applySdkEnvironmentOverrides(pipeline, options), equalTo(expectedPipeline));
  }