static UnboundedCountingSource createUnboundedFrom(long start) {
    return new UnboundedCountingSource(start, 1, 1L, Duration.ZERO, new NowTimestampFn());
  }
@Test
  @Category(NeedsRunner.class)
  public void testUnboundedSourceWithRate() {

    Duration period = Duration.millis(5);
    long numElements = 1000L;

    PCollection<Long> input =
        p.apply(
            Read.from(
                    CountingSource.createUnboundedFrom(0)
                        .withTimestampFn(new ValueAsTimestampFn())
                        .withRate(1, period))
                .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);

    Instant started = Instant.now();
    p.run();
    Instant finished = Instant.now();
    Duration expectedDuration = period.multipliedBy((int) numElements);
    assertThat(started.plus(expectedDuration).isBefore(finished), is(true));
  }