static UnboundedCountingSource createUnboundedFrom(long start) {
    return new UnboundedCountingSource(start, 1, 1L, Duration.ZERO, new NowTimestampFn());
  }
@Test
  @Category(NeedsRunner.class)
  public void testUnboundedSourceRateSplits() throws Exception {
    int elementsPerPeriod = 10;
    Duration period = Duration.millis(5);

    long numElements = 1000;
    int numSplits = 10;

    UnboundedCountingSource initial =
        CountingSource.createUnboundedFrom(0).withRate(elementsPerPeriod, period);
    List<? extends UnboundedSource<Long, ?>> splits = initial.split(numSplits, p.getOptions());
    assertEquals("Expected exact splitting", numSplits, splits.size());

    long elementsPerSplit = numElements / numSplits;
    assertEquals("Expected even splits", numElements, elementsPerSplit * numSplits);
    PCollectionList<Long> pcollections = PCollectionList.empty(p);
    for (int i = 0; i < splits.size(); ++i) {
      pcollections =
          pcollections.and(
              p.apply("split" + i, Read.from(splits.get(i)).withMaxNumRecords(elementsPerSplit)));
    }
    PCollection<Long> input = pcollections.apply(Flatten.pCollections());

    addCountingAsserts(input, numElements);
    Instant startTime = Instant.now();
    p.run();
    Instant endTime = Instant.now();
    // 500 ms if the readers are all initialized in parallel; 5000 ms if they are evaluated serially
    long expectedMinimumMillis = (numElements * period.getMillis()) / elementsPerPeriod;
    assertThat(expectedMinimumMillis, lessThan(endTime.getMillis() - startTime.getMillis()));
  }