public static <T> Bounded<T> from(BoundedSource<T> source) {
    return new Bounded<>(null, source);
  }
@Test
  @Category({
    NeedsRunner.class,
    UsesUnboundedPCollections.class,
    UsesUnboundedSplittableParDo.class
  })
  public void testUnboundedSdfWrapperCacheStartedReaders() {
    long numElements = 1000L;
    PCollection<Long> input =
        pipeline.apply(Read.from(new ExpectCacheUnboundedSource(numElements)));
    PAssert.that(input)
        .containsInAnyOrder(
            LongStream.rangeClosed(1L, numElements).boxed().collect(Collectors.toList()));
    // TODO(https://github.com/apache/beam/issues/20530): Remove additional experiments when SDF
    // read is default.
    ExperimentalOptions.addExperiment(
        pipeline.getOptions().as(ExperimentalOptions.class), "use_sdf_read");
    // Force the pipeline to run with one thread to ensure the reader will be reused on one DoFn
    // instance.
    // We are not able to use DirectOptions because of circular dependency.
    pipeline
        .runWithAdditionalOptionArgs(ImmutableList.of("--targetParallelism=1"))
        .waitUntilFinish();
  }