@Override
  public List<SyntheticUnboundedSource> split(int desiredNumSplits, PipelineOptions options) {
    int desiredNumBundles =
        sourceOptions.forceNumInitialBundles != null
            ? sourceOptions.forceNumInitialBundles
            : desiredNumSplits;

    List<SyntheticUnboundedSource> splits =
        bundleSplitter.getBundleSizes(desiredNumBundles, startOffset, endOffset).stream()
            .map(
                offsetRange ->
                    new SyntheticUnboundedSource(
                        offsetRange.getFrom(), offsetRange.getTo(), sourceOptions))
            .collect(Collectors.toList());
    LOG.info("Split into {} bundles: {}", splits.size(), splits);

    return splits;
  }
@Test
  public void shouldCreateSplitsOfCountProvidedInOptions() throws IOException {
    sourceOptions.forceNumInitialBundles = 20;

    source = new SyntheticUnboundedSource(sourceOptions);

    // desiredNumSplits should be ignored if we specify forceNumInitialBundles
    Integer splitCount = source.split(100000, pipeline.getOptions()).size();

    assertEquals(sourceOptions.forceNumInitialBundles, splitCount);
  }