PeriodicSequence extends PTransform<PCollection<PeriodicSequence.SequenceDefinition>, PCollection<Instant>> { public static PeriodicSequence create() { return new PeriodicSequence(); } private  PeriodicSequence(); static PeriodicSequence create(); @Override PCollection<Instant> expand(PCollection<SequenceDefinition> input);  }
@Test @Category({ NeedsRunner.class, UsesImpulse.class, UsesStatefulParDo.class, }) public void testOutputsProperElements() { Instant instant = Instant.now(); Instant startTime = instant.minus(Duration.standardHours(100)); long duration = 500; Duration interval = Duration.millis(250); long intervalMillis = interval.getMillis(); Instant stopTime = startTime.plus(duration); PCollection<KV<Instant, Instant>> result = p.apply( Create.<PeriodicSequence.SequenceDefinition>of( new PeriodicSequence.SequenceDefinition(startTime, stopTime, interval))) .apply(PeriodicSequence.create()) .apply(ParDo.of(new ExtractTsDoFn<>())); ArrayList<KV<Instant, Instant>> expectedResults = new ArrayList<>((int) (duration / intervalMillis + 1)); for (long i = 0; i <= duration; i += intervalMillis) { Instant el = startTime.plus(i); expectedResults.add(KV.of(el, el)); } PAssert.that(result).containsInAnyOrder(expectedResults); p.run().waitUntilFinish(); }