SyntheticBoundedSource extends OffsetBasedSource<KV<byte[], byte[]>> { @Override public List<SyntheticBoundedSource> split(long desiredBundleSizeBytes, PipelineOptions options) throws Exception { 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; } SyntheticBoundedSource(SyntheticSourceOptions sourceOptions);  SyntheticBoundedSource(
      long startOffset, long endOffset, SyntheticSourceOptions sourceOptions); @Override Coder<KV<byte[], byte[]>> getDefaultOutputCoder(); @Override // TODO: test cases where the source size could not be estimated (i.e., return 0). // TODO: test cases where the key size and value size might differ from record to record. // The key size and value size might have their own distributions. long getBytesPerOffset(); @Override void validate(); @Override String toString(); @Override final SyntheticBoundedSource createSourceForSubrange(long start, long end); @Override long getMaxEndOffset(PipelineOptions options); @Override SyntheticSourceReader createReader(PipelineOptions pipelineOptions); @Override List<SyntheticBoundedSource> split(long desiredBundleSizeBytes, PipelineOptions options);  }
@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()); }