@Deprecated
  public static BoundedSource<Long> upTo(long numElements) {
    checkArgument(
        numElements >= 0, "numElements (%s) must be greater than or equal to 0", numElements);
    return new BoundedCountingSource(0, numElements);
  }
@Test
  @Category({
    ValidatesRunner.class,
    UsesStatefulParDo.class // This test fails if State is unsupported despite no direct usage.
  })
  public void testBoundedSourceSplits() throws Exception {
    long numElements = 1000;
    long numSplits = 10;
    long splitSizeBytes = numElements * 8 / numSplits; // 8 bytes per long element.

    BoundedSource<Long> initial = CountingSource.upTo(numElements);
    List<? extends BoundedSource<Long>> splits = initial.split(splitSizeBytes, p.getOptions());
    assertEquals("Expected exact splitting", numSplits, splits.size());

    // Assemble all the splits into one flattened PCollection, also verify their sizes.
    PCollectionList<Long> pcollections = PCollectionList.empty(p);
    for (int i = 0; i < splits.size(); ++i) {
      BoundedSource<Long> split = splits.get(i);
      pcollections = pcollections.and(p.apply("split" + i, Read.from(split)));
      assertEquals(
          "Expected even splitting", splitSizeBytes, split.getEstimatedSizeBytes(p.getOptions()));
    }
    PCollection<Long> input = pcollections.apply(Flatten.pCollections());

    addCountingAsserts(input, numElements);
    p.run();
  }