AvroByteReader extends NativeReader<T> { @Override public AvroByteFileIterator iterator() throws IOException { BoundedSource.BoundedReader<ByteBuffer> reader; if (startPosition == 0 && endPosition == Long.MAX_VALUE) { reader = avroSource.createReader(options); } else { reader = avroSource .createForSubrangeOfFile( FileSystems.matchSingleFileSpec(filename), startPosition, endPosition) .createReader(options); } return new AvroByteFileIterator((AvroReader<ByteBuffer>) reader); } AvroByteReader(
      String filename,
      long startPosition,
      long endPosition,
      Coder<T> coder,
      PipelineOptions options); @Override AvroByteFileIterator iterator();  }
@Test public void testGetProgress() throws Exception { List<List<String>> elements = generateInputBlocks(4, 10, 10); Coder<String> coder = StringUtf8Coder.of(); AvroFileInfo<String> fileInfo = initInputFile(elements, coder); AvroByteReader<String> reader = new AvroByteReader<String>( fileInfo.filename, 0L, Long.MAX_VALUE, coder, PipelineOptionsFactory.create()); AvroByteFileIterator iterator = reader.iterator(); Assert.assertTrue(iterator.start()); ApproximateReportedProgress progress = readerProgressToCloudProgress(reader.iterator().getProgress()); Assert.assertEquals(0.0, progress.getConsumedParallelism().getValue(), 1e-6); Assert.assertEquals(0.0, progress.getFractionConsumed(), 1e-6); Assert.assertNull(progress.getRemainingParallelism()); Assert.assertTrue(iterator.advance()); progress = readerProgressToCloudProgress(iterator.getProgress()); Assert.assertEquals(1.0, progress.getConsumedParallelism().getValue(), 1e-6); Assert.assertNull(progress.getRemainingParallelism()); Assert.assertTrue(iterator.advance()); progress = readerProgressToCloudProgress(iterator.getProgress()); Assert.assertEquals(2.0, progress.getConsumedParallelism().getValue(), 1e-6); Assert.assertNull(progress.getRemainingParallelism()); Assert.assertTrue(iterator.advance()); progress = readerProgressToCloudProgress(iterator.getProgress()); Assert.assertEquals(3.0, progress.getConsumedParallelism().getValue(), 1e-6); Assert.assertEquals(1.0, progress.getRemainingParallelism().getValue(), 1e-6); Assert.assertFalse(iterator.advance()); progress = readerProgressToCloudProgress(iterator.getProgress()); Assert.assertEquals(4.0, progress.getConsumedParallelism().getValue(), 1e-6); Assert.assertEquals(0.0, progress.getRemainingParallelism().getValue(), 1e-6); Assert.assertEquals(1.0, progress.getFractionConsumed(), 1e-6); }