@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 givenUnboundedSource_runsUntilCancel() throws IOException {
    Pipeline pipeline = Pipeline.create(options());
    pipeline.apply(PeriodicImpulse.create());
    PipelineResult result = pipeline.run();
    assertThat(result.getState()).isEqualTo(PipelineResult.State.RUNNING);
    PipelineResult.State state = result.cancel();
    assertThat(state).isEqualTo(PipelineResult.State.CANCELLED);
  }