SyntheticBoundedSource extends OffsetBasedSource<KV<byte[], byte[]>> { @Override public SyntheticSourceReader createReader(PipelineOptions pipelineOptions) { return new SyntheticSourceReader(this); } 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 testIncreasingProgress() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); testSourceOptions.progressShape = ProgressShape.LINEAR; SyntheticBoundedSource source = new SyntheticBoundedSource(testSourceOptions); BoundedSource.BoundedReader<KV<byte[], byte[]>> reader = source.createReader(options); assertEquals(0, reader.getFractionConsumed(), 1e-5); double lastFractionConsumed = -1.0; for (boolean more = reader.start(); more; more = reader.advance()) { assertTrue(reader.getFractionConsumed() > lastFractionConsumed); lastFractionConsumed = reader.getFractionConsumed(); } assertEquals(1, reader.getFractionConsumed(), 1e-5); }
@Test public void testRegressingProgress() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); testSourceOptions.progressShape = ProgressShape.LINEAR_REGRESSING; SyntheticBoundedSource source = new SyntheticBoundedSource(testSourceOptions); BoundedSource.BoundedReader<KV<byte[], byte[]>> reader = source.createReader(options); double lastFractionConsumed = reader.getFractionConsumed(); for (boolean more = reader.start(); more; more = reader.advance()) { assertTrue(reader.getFractionConsumed() <= lastFractionConsumed); lastFractionConsumed = reader.getFractionConsumed(); } }