@Override
  public InMemoryReaderIterator iterator() throws IOException {
    return new InMemoryReaderIterator();
  }
@Test
  public void testProgressReporting() throws Exception {
    List<Integer> elements = Arrays.asList(33, 44, 55, 66, 77, 88);
    // Should initially read elements at indices: 44@1, 55@2, 66@3, 77@4

    Coder<Integer> coder = BigEndianIntegerCoder.of();
    InMemoryReader<Integer> inMemoryReader =
        new InMemoryReader<>(encodedElements(elements, coder), 1, 4, coder);
    try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) {
      assertNull(iterator.getProgress());
      assertEquals(3, iterator.getRemainingParallelism(), 0.0);

      assertTrue(iterator.start());
      Assert.assertEquals(
          ReaderTestUtils.positionAtIndex(1L),
          ReaderTestUtils.positionFromProgress(iterator.getProgress()));
      assertEquals(3, iterator.getRemainingParallelism(), 0.0);

      assertTrue(iterator.advance());
      Assert.assertEquals(
          ReaderTestUtils.positionAtIndex(2L),
          ReaderTestUtils.positionFromProgress(iterator.getProgress()));
      assertEquals(2, iterator.getRemainingParallelism(), 0.0);

      assertTrue(iterator.advance());
      Assert.assertEquals(
          ReaderTestUtils.positionAtIndex(3L),
          ReaderTestUtils.positionFromProgress(iterator.getProgress()));
      assertEquals(1, iterator.getRemainingParallelism(), 0.0);

      assertFalse(iterator.advance());
    }
  }