@Override
  public InMemoryReaderIterator iterator() throws IOException {
    return new InMemoryReaderIterator();
  }
@Test
  public void testDynamicSplit() 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);

    // Unstarted iterator.
    try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) {
      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L)));
    }

    // Illegal proposed split position.
    try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) {
      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L)));
      // Poke the iterator so that we can test dynamic splitting.
      assertTrue(iterator.start());
      assertNull(
          iterator.requestDynamicSplit(toDynamicSplitRequest(new ApproximateSplitRequest())));
      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(null)));
    }

    // Successful update.
    try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) {
      // Poke the iterator so that we can test dynamic splitting.
      assertTrue(iterator.start());
      NativeReader.DynamicSplitResult dynamicSplitResult =
          iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L));
      Assert.assertEquals(
          ReaderTestUtils.positionAtIndex(3L),
          ReaderTestUtils.positionFromSplitResult(dynamicSplitResult));
      assertEquals(3, iterator.tracker.getStopPosition().longValue());
      assertEquals(44, iterator.getCurrent().intValue());
      assertTrue(iterator.advance());
      assertEquals(55, iterator.getCurrent().intValue());
      assertFalse(iterator.advance());
    }

    // Proposed split position is before the current position, no update.
    try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) {
      // Poke the iterator so that we can test dynamic splitting.
      assertTrue(iterator.start());
      assertEquals(44, iterator.getCurrent().intValue());
      assertTrue(iterator.advance());
      assertEquals(55, iterator.getCurrent().intValue());
      assertTrue(iterator.advance()); // Returns true => we promised to return 66.
      // Now we have to refuse the split.
      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(3L)));
      assertEquals(4, iterator.tracker.getStopPosition().longValue());
      assertEquals(66, iterator.getCurrent().intValue());
      assertFalse(iterator.advance());
    }

    // Proposed split position is after the current stop (end) position, no update.
    try (InMemoryReader<Integer>.InMemoryReaderIterator iterator = inMemoryReader.iterator()) {
      // Poke the iterator so that we can test dynamic splitting.
      assertTrue(iterator.start());
      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtIndex(5L)));
      assertEquals(4, iterator.tracker.getStopPosition().longValue());
    }
  }