public static <T, IdT> Deduplicate.WithRepresentativeValues<T, IdT> withRepresentativeValueFn(
      SerializableFunction<T, IdT> representativeValueFn) {
    return new Deduplicate.WithRepresentativeValues<T, IdT>(
        DEFAULT_TIME_DOMAIN, DEFAULT_DURATION, representativeValueFn, null, null);
  }
@Test
  @Category({NeedsRunner.class, UsesTestStreamWithProcessingTime.class})
  public void testRepresentativeValuesWithCoder() {
    Instant base = new Instant(0);
    TestStream<KV<Long, String>> values =
        TestStream.create(KvCoder.of(VarLongCoder.of(), StringUtf8Coder.of()))
            .advanceWatermarkTo(base)
            .addElements(
                TimestampedValue.of(KV.of(1L, "k1"), base),
                TimestampedValue.of(KV.of(2L, "k2"), base.plus(Duration.standardSeconds(10))),
                TimestampedValue.of(KV.of(3L, "k3"), base.plus(Duration.standardSeconds(20))))
            .advanceProcessingTime(Duration.standardMinutes(1))
            .addElements(
                TimestampedValue.of(KV.of(1L, "k1"), base.plus(Duration.standardSeconds(30))),
                TimestampedValue.of(KV.of(2L, "k2"), base.plus(Duration.standardSeconds(40))),
                TimestampedValue.of(KV.of(3L, "k3"), base.plus(Duration.standardSeconds(50))))
            .advanceWatermarkToInfinity();

    PCollection<KV<Long, String>> distinctValues =
        p.apply(values)
            .apply(
                Deduplicate.withRepresentativeValueFn(new Keys<Long>())
                    .withRepresentativeCoder(VarLongCoder.of()));

    PAssert.that(distinctValues)
        .containsInAnyOrder(KV.of(1L, "k1"), KV.of(2L, "k2"), KV.of(3L, "k3"));
    p.run();
  }