public static PeriodicSequence create() {
    return new PeriodicSequence();
  }
@Test
  @Category({
    NeedsRunner.class,
    UsesImpulse.class,
    UsesStatefulParDo.class,
    UsesUnboundedPCollections.class,
    UsesUnboundedSplittableParDo.class
  })
  public void testOutputsProperElements() {
    Instant startTime = Instant.now().plus(Duration.standardSeconds(2));
    Duration interval = Duration.millis(250);
    long intervalMillis = interval.getMillis();
    long duration = 3 * intervalMillis;
    Instant stopTime = startTime.plus(Duration.millis(duration));

    PCollection<TimestampedValue<KV<Instant, Instant>>> result =
        p.apply(Create.of(new PeriodicSequence.SequenceDefinition(startTime, stopTime, interval)))
            .apply(PeriodicSequence.create())
            .apply(
                Window.<Instant>into(FixedWindows.of(interval))
                    .withTimestampCombiner(TimestampCombiner.EARLIEST))
            .apply(WithKeys.of("dummy"))
            .apply(GroupByKey.create())
            .apply(
                MapElements.into(TypeDescriptor.of(Instant.class))
                    .via(e -> Iterables.getOnlyElement(e.getValue())))
            .apply(ParDo.of(new ExtractTsDoFn<>())); // validate timestamps

    ArrayList<Instant> expectedResults = new ArrayList<>();
    for (long i = 0; i <= duration; i += intervalMillis) {
      Instant el = startTime.plus(Duration.millis(i));
      expectedResults.add(el);
    }

    PAssert.that(result)
        .satisfies(
            values -> {
              List<TimestampedValue<KV<Instant, Instant>>> sortedValues =
                  Streams.stream(values)
                      .sorted(Comparator.comparing(e -> e.getValue().getValue()))
                      .collect(Collectors.toList());

              assertEquals(
                  expectedResults,
                  sortedValues.stream()
                      .map(e -> e.getValue().getValue())
                      .collect(Collectors.toList()));

              Instant minTs =
                  sortedValues.stream()
                      .min(Comparator.comparing(TimestampedValue::getTimestamp))
                      .get()
                      .getTimestamp();
              Instant maxTs =
                  sortedValues.stream()
                      .max(Comparator.comparing(TimestampedValue::getTimestamp))
                      .get()
                      .getTimestamp();
              final long expectedDiff = intervalMillis / 2;
              assertTrue(
                  String.format(
                      "Expected processing-time diff at least %d, got %d",
                      expectedDiff, maxTs.getMillis() - minTs.getMillis()),
                  maxTs.getMillis() - minTs.getMillis() > expectedDiff);
              return null;
            });

    p.run().waitUntilFinish();
  }