@Override
  public PipelineResult run(Pipeline pipeline) {
    LOG.info(
        "running Pipeline using {}: defaultEnvironmentType: {}, jobEndpoint: {}",
        PortableRunner.class.getName(),
        prismPipelineOptions.getDefaultEnvironmentType(),
        prismPipelineOptions.getJobEndpoint());

    return internal.run(pipeline);
  }
@Test
  public void givenBoundedSource_runsUntilDone() {
    Pipeline pipeline = Pipeline.create(options());
    pipeline.apply(Create.of(1, 2, 3));
    PipelineResult.State state = pipeline.run().waitUntilFinish();
    assertThat(state).isEqualTo(PipelineResult.State.DONE);
  }