SyntheticUnboundedSource extends UnboundedSource<KV<byte[], byte[]>, SyntheticRecordsCheckpoint> { @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 of sizes: {}", splits.size(), splits); return splits; } SyntheticUnboundedSource(SyntheticSourceOptions sourceOptions); private  SyntheticUnboundedSource(
      long startOffset, long endOffset, SyntheticSourceOptions sourceOptions); @Override Coder<KV<byte[], byte[]>> getOutputCoder(); @Override void validate(); @Override Coder<SyntheticRecordsCheckpoint> getCheckpointMarkCoder(); @Override UnboundedReader<KV<byte[], byte[]>> createReader(
      PipelineOptions options, @Nullable SyntheticRecordsCheckpoint checkpoint); @Override List<SyntheticUnboundedSource> split(int desiredNumSplits, PipelineOptions options);  }
@Test public void shouldCreateSplitsOfCountProvidedInOptions() throws IOException { sourceOptions.forceNumInitialBundles = 20; source = new SyntheticUnboundedSource(sourceOptions); Integer splitCount = source.split(100000, pipeline.getOptions()).size(); assertEquals(sourceOptions.forceNumInitialBundles, splitCount); }