OffsetBasedSource extends BoundedSource<T> { @Override public String toString() { return "[" + startOffset + ", " + endOffset + ")"; } OffsetBasedSource(long startOffset, long endOffset, long minBundleSize); long getStartOffset(); long getEndOffset(); long getMinBundleSize(); @Override long getEstimatedSizeBytes(PipelineOptions options); @Override List<? extends OffsetBasedSource<T>> split(
      long desiredBundleSizeBytes, PipelineOptions options); @Override void validate(); @Override String toString(); long getBytesPerOffset(); abstract long getMaxEndOffset(PipelineOptions options); abstract OffsetBasedSource<T> createSourceForSubrange(long start, long end); @Override void populateDisplayData(DisplayData.Builder builder);  }
@Test public void testReadingGranularityAndFractionConsumed() throws IOException { PipelineOptions options = PipelineOptionsFactory.create(); CoarseRangeSource source = new CoarseRangeSource(13, 35, 1, 10); try (CoarseRangeReader reader = source.createReader(options)) { List<Integer> items = new ArrayList<>(); assertEquals(0.0, reader.getFractionConsumed(), 1e-6); assertTrue(reader.start()); items.add(reader.getCurrent()); while (reader.advance()) { Double fraction = reader.getFractionConsumed(); assertNotNull(fraction); assertTrue(fraction.toString(), fraction > 0.0); assertTrue(fraction.toString(), fraction <= 1.0); items.add(reader.getCurrent()); } assertEquals(1.0, reader.getFractionConsumed(), 1e-6); assertEquals(20, items.size()); assertEquals(20, items.get(0).intValue()); assertEquals(39, items.get(items.size() - 1).intValue()); source = new CoarseRangeSource(13, 17, 1, 10); } try (BoundedSource.BoundedReader<Integer> reader = source.createReader(options)) { assertFalse(reader.start()); } }