[Bug]: Periodic Impulse does not drain on Dataflow

[Abacn]

What happened?

Draining a pipeline with source from PeriodicImpulse never ends. This happens for both Java and Python sdk.
In contrast, similar functionality GenericSequence which existing in Java supports drain.

Code snippets:

public class PeriodicImpulseTest {
  public static void main(String[] argv) {
    PipelineOptions options = PipelineOptionsFactory.fromArgs(argv).withValidation().as(PipelineOptions.class);
    Pipeline p = Pipeline.create(options);

    PCollection<Void> result =
        p.apply(PeriodicImpulse.create().withInterval(Duration.millis(500)))
            .apply(Reshuffle.viaRandomKey())
            .apply(ParDo.of(new DoFn<Instant, Void>() {
              @ProcessElement
              public void processElement(DoFn<Instant, Void>.ProcessContext c){
                System.out.println(c.element());
              }}
            ));
    assertThat(result.isBounded(), equalTo(IsBounded.UNBOUNDED));
    p.run().waitUntilFinish();
  }
}

Successful implementation using GenerateSequence:

public class GenerateSequenceTest {
  public static void main(String[] argv) {
    PipelineOptions options = PipelineOptionsFactory.fromArgs(argv).withValidation().as(PipelineOptions.class);
    Pipeline p = Pipeline.create(options);

    PCollection<Long> input = p.apply(GenerateSequence.from(0).withRate(1,
        Duration.standardSeconds(1)));
    input.apply(ParDo.of(new DoFn<Long, Void>() {
        @ProcessElement
        public void processElement(DoFn<Long, Void>.ProcessContext c){
          System.out.println(c.element());
        }}
    ));
    p.run();
  }
}

Issue Priority

Priority: 2

Issue Component

Component: sdk-java-core

[Abacn]

.add-labels python

[Abacn]

Python SDK issue resolved. May keep this open for Java SDK.

[damccorm]

Oops - forgot this was happening both places