public static <T> Deduplicate.Values<T> values() {
    return new Deduplicate.Values<>(DEFAULT_TIME_DOMAIN, DEFAULT_DURATION);
  }
@Test
  @Category({NeedsRunner.class, UsesTestStream.class})
  public void testInDifferentWindows() {
    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("k1", base.plus(Duration.standardSeconds(30))),
                TimestampedValue.of("k2", base.plus(Duration.standardSeconds(40))),
                TimestampedValue.of("k3", base.plus(Duration.standardSeconds(50))),
                TimestampedValue.of("k4", base.plus(Duration.standardSeconds(60))),
                TimestampedValue.of("k5", base.plus(Duration.standardSeconds(70))),
                TimestampedValue.of("k6", base.plus(Duration.standardSeconds(80))))
            .advanceWatermarkToInfinity();

    PCollection<String> distinctValues =
        p.apply(values)
            .apply(Window.into(FixedWindows.of(Duration.standardSeconds(30))))
            .apply(Deduplicate.values());
    PAssert.that(distinctValues)
        .inWindow(new IntervalWindow(base, base.plus(Duration.standardSeconds(30))))
        .containsInAnyOrder("k1", "k2", "k3");
    PAssert.that(distinctValues)
        .inWindow(
            new IntervalWindow(
                base.plus(Duration.standardSeconds(30)), base.plus(Duration.standardSeconds(60))))
        .containsInAnyOrder("k1", "k2", "k3");
    PAssert.that(distinctValues)
        .inWindow(
            new IntervalWindow(
                base.plus(Duration.standardSeconds(60)), base.plus(Duration.standardSeconds(90))))
        .containsInAnyOrder("k4", "k5", "k6");
    p.run();
  }