public static <T> Distinct<T> create() {
    return new Distinct<>();
  }
@Test
  @Category({NeedsRunner.class, UsesTestStreamWithProcessingTime.class})
  public void testTriggeredDistinct() {
    Instant base = new Instant(0);
    TestStream<String> values =
        TestStream.create(StringUtf8Coder.of())
            .advanceWatermarkTo(base)
            .addElements(
                TimestampedValue.of("k1", base),
                TimestampedValue.of("k2", base.plus(Duration.standardSeconds(10))),
                TimestampedValue.of("k3", base.plus(Duration.standardSeconds(20))))
            .advanceProcessingTime(Duration.standardMinutes(1))
            .addElements(
                TimestampedValue.of("k1", base.plus(Duration.standardSeconds(30))),
                TimestampedValue.of("k2", base.plus(Duration.standardSeconds(40))),
                TimestampedValue.of("k3", base.plus(Duration.standardSeconds(50))))
            .advanceWatermarkToInfinity();

    PCollection<String> distinctValues =
        triggeredDistinctPipeline
            .apply(values)
            .apply(
                Window.<String>into(FixedWindows.of(Duration.standardMinutes(1)))
                    .triggering(
                        Repeatedly.forever(
                            AfterProcessingTime.pastFirstElementInPane()
                                .plusDelayOf(Duration.standardSeconds(30))))
                    .withAllowedLateness(Duration.ZERO)
                    .accumulatingFiredPanes())
            .apply(Distinct.create());
    PAssert.that(distinctValues).containsInAnyOrder("k1", "k2", "k3");
    triggeredDistinctPipeline.run();
  }