@Override
  public SyntheticSourceReader createReader(PipelineOptions pipelineOptions) {
    return new SyntheticSourceReader(this);
  }
@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();
    }
  }