@Override
  public List<SyntheticBoundedSource> split(long desiredBundleSizeBytes, PipelineOptions options)
      throws Exception {
    // Choose number of bundles either based on explicit parameter,
    // or based on size and hints.
    int desiredNumBundles =
        (sourceOptions.forceNumInitialBundles == null)
            ? ((int) Math.ceil(1.0 * getEstimatedSizeBytes(options) / desiredBundleSizeBytes))
            : sourceOptions.forceNumInitialBundles;

    List<SyntheticBoundedSource> res =
        bundleSplitter.getBundleSizes(desiredNumBundles, this.getStartOffset(), this.getEndOffset())
            .stream()
            .map(offsetRange -> createSourceForSubrange(offsetRange.getFrom(), offsetRange.getTo()))
            .collect(Collectors.toList());
    LOG.info("Split into {} bundles of sizes: {}", res.size(), res);
    return res;
  }
@Test
  public void testSplitIntoBundles() throws Exception {
    testSplitIntoBundlesP(1);
    testSplitIntoBundlesP(-1);
    testSplitIntoBundlesP(5);

    PipelineOptions options = PipelineOptionsFactory.create();
    testSourceOptions.forceNumInitialBundles = 37;
    assertEquals(37, new SyntheticBoundedSource(testSourceOptions).split(42, options).size());
  }