@VisibleForTesting
  protected RunnerApi.Pipeline resolveArtifacts(RunnerApi.Pipeline pipeline) {
    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();
      environmentBuilder.clearDependencies();
      for (RunnerApi.ArtifactInformation info : entry.getValue().getDependenciesList()) {
        if (!BeamUrns.getUrn(RunnerApi.StandardArtifacts.Types.FILE).equals(info.getTypeUrn())) {
          throw new RuntimeException(
              String.format("unsupported artifact type %s", info.getTypeUrn()));
        }
        RunnerApi.ArtifactFilePayload filePayload;
        try {
          filePayload = RunnerApi.ArtifactFilePayload.parseFrom(info.getTypePayload());
        } catch (InvalidProtocolBufferException e) {
          throw new RuntimeException("Error parsing artifact file payload.", e);
        }
        String stagedName;
        if (BeamUrns.getUrn(RunnerApi.StandardArtifacts.Roles.STAGING_TO)
            .equals(info.getRoleUrn())) {
          try {
            RunnerApi.ArtifactStagingToRolePayload stagingPayload =
                RunnerApi.ArtifactStagingToRolePayload.parseFrom(info.getRolePayload());
            stagedName = stagingPayload.getStagedName();
          } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException("Error parsing artifact staging_to role payload.", e);
          }
        } else {
          try {
            File source = new File(filePayload.getPath());
            HashCode hashCode = Files.asByteSource(source).hash(Hashing.sha256());
            stagedName = Environments.createStagingFileName(source, hashCode);
          } catch (IOException e) {
            throw new RuntimeException(
                String.format("Error creating staged name for artifact %s", filePayload.getPath()),
                e);
          }
        }
        environmentBuilder.addDependencies(
            info.toBuilder()
                .setTypeUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Types.URL))
                .setTypePayload(
                    RunnerApi.ArtifactUrlPayload.newBuilder()
                        .setUrl(
                            FileSystems.matchNewResource(options.getStagingLocation(), true)
                                .resolve(
                                    stagedName, ResolveOptions.StandardResolveOptions.RESOLVE_FILE)
                                .toString())
                        .setSha256(filePayload.getSha256())
                        .build()
                        .toByteString()));
      }
      componentsBuilder.putEnvironments(entry.getKey(), environmentBuilder.build());
    }
    return pipelineBuilder.build();
  }
@Test
  public void testResolveArtifacts() throws IOException {
    DataflowPipelineOptions options = buildPipelineOptions();
    DataflowRunner runner = DataflowRunner.fromOptions(options);
    String stagingLocation = options.getStagingLocation().replaceFirst("/$", "");
    RunnerApi.ArtifactInformation fooLocalArtifact =
        RunnerApi.ArtifactInformation.newBuilder()
            .setTypeUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Types.FILE))
            .setTypePayload(
                RunnerApi.ArtifactFilePayload.newBuilder()
                    .setPath("/tmp/foo.jar")
                    .build()
                    .toByteString())
            .setRoleUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Roles.STAGING_TO))
            .setRolePayload(
                RunnerApi.ArtifactStagingToRolePayload.newBuilder()
                    .setStagedName("foo_staged.jar")
                    .build()
                    .toByteString())
            .build();
    RunnerApi.ArtifactInformation barLocalArtifact =
        RunnerApi.ArtifactInformation.newBuilder()
            .setTypeUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Types.FILE))
            .setTypePayload(
                RunnerApi.ArtifactFilePayload.newBuilder()
                    .setPath("/tmp/bar.jar")
                    .build()
                    .toByteString())
            .setRoleUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Roles.STAGING_TO))
            .setRolePayload(
                RunnerApi.ArtifactStagingToRolePayload.newBuilder()
                    .setStagedName("bar_staged.jar")
                    .build()
                    .toByteString())
            .build();
    RunnerApi.Pipeline pipeline =
        RunnerApi.Pipeline.newBuilder()
            .setComponents(
                RunnerApi.Components.newBuilder()
                    .putEnvironments(
                        "env",
                        RunnerApi.Environment.newBuilder()
                            .addAllDependencies(
                                ImmutableList.of(fooLocalArtifact, barLocalArtifact))
                            .build()))
            .build();

    RunnerApi.ArtifactInformation fooStagedArtifact =
        RunnerApi.ArtifactInformation.newBuilder()
            .setTypeUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Types.URL))
            .setTypePayload(
                RunnerApi.ArtifactUrlPayload.newBuilder()
                    .setUrl(stagingLocation + "/foo_staged.jar")
                    .build()
                    .toByteString())
            .setRoleUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Roles.STAGING_TO))
            .setRolePayload(
                RunnerApi.ArtifactStagingToRolePayload.newBuilder()
                    .setStagedName("foo_staged.jar")
                    .build()
                    .toByteString())
            .build();
    RunnerApi.ArtifactInformation barStagedArtifact =
        RunnerApi.ArtifactInformation.newBuilder()
            .setTypeUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Types.URL))
            .setTypePayload(
                RunnerApi.ArtifactUrlPayload.newBuilder()
                    .setUrl(stagingLocation + "/bar_staged.jar")
                    .build()
                    .toByteString())
            .setRoleUrn(BeamUrns.getUrn(RunnerApi.StandardArtifacts.Roles.STAGING_TO))
            .setRolePayload(
                RunnerApi.ArtifactStagingToRolePayload.newBuilder()
                    .setStagedName("bar_staged.jar")
                    .build()
                    .toByteString())
            .build();
    RunnerApi.Pipeline expectedPipeline =
        RunnerApi.Pipeline.newBuilder()
            .setComponents(
                RunnerApi.Components.newBuilder()
                    .putEnvironments(
                        "env",
                        RunnerApi.Environment.newBuilder()
                            .addAllDependencies(
                                ImmutableList.of(fooStagedArtifact, barStagedArtifact))
                            .build()))
            .build();
    assertThat(runner.resolveArtifacts(pipeline), equalTo(expectedPipeline));
  }