public static <T> Deduplicate.Values<T> values() {
    return new Deduplicate.Values<>(DEFAULT_TIME_DOMAIN, DEFAULT_DURATION);
  }
@Test
  @Category({NeedsRunner.class, UsesTestStreamWithProcessingTime.class})
  public void testProcessingTime() {
    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))),
                TimestampedValue.of("maybedup", base.plus(Duration.standardSeconds(59))))
            .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))))
            .advanceProcessingTime(Deduplicate.DEFAULT_DURATION)
            .addElements(TimestampedValue.of("maybedup", base.plus(Duration.standardSeconds(59))))
            .advanceWatermarkToInfinity();

    PCollection<String> distinctValues = p.apply(values).apply(Deduplicate.values());
    PAssert.that(distinctValues)
        .satisfies(
            (Iterable<String> input) -> {
              assertEquals(1, Iterables.frequency(input, "k1"));
              assertEquals(1, Iterables.frequency(input, "k2"));
              assertEquals(1, Iterables.frequency(input, "k3"));
              assertTrue(
                  Iterables.frequency(input, "maybedup") == 1
                      || Iterables.frequency(input, "maybedup") == 2);
              return null;
            });
    p.run();
  }