@Deprecated
  public static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn(
      SerializableFunction<Long, Instant> timestampFn) {
    return new UnboundedCountingSource(0, 1, 1L, Duration.ZERO, timestampFn);
  }
@Test
  @Category(NeedsRunner.class)
  public void testUnboundedSourceTimestamps() {
    long numElements = 1000;

    PCollection<Long> input =
        p.apply(
            Read.from(CountingSource.unboundedWithTimestampFn(new ValueAsTimestampFn()))
                .withMaxNumRecords(numElements));
    addCountingAsserts(input, numElements);

    PCollection<Long> diffs =
        input
            .apply("TimestampDiff", ParDo.of(new ElementValueDiff()))
            .apply("DistinctTimestamps", Distinct.create());
    // This assert also confirms that diffs only has one unique value.
    PAssert.thatSingleton(diffs).isEqualTo(0L);

    p.run();
  }