@Test
  public void testReplicationSourceWALReader() throws Exception {
    appendEntriesToLogAndSync(3);
    // get ending position
    long position;
    try (WALEntryStream entryStream =
      new WALEntryStream(logQueue, fs, CONF, 0, log, new MetricsSource("1"), fakeWalGroupId)) {
      for (int i = 0; i < 3; i++) {
        assertNotNull(next(entryStream));
      }
      position = entryStream.getPosition();
    }

    // start up a reader
    Path walPath = getQueue().peek();
    ReplicationSourceWALReader reader = createReader(false, CONF);
    WALEntryBatch entryBatch = reader.take();

    // should've batched up our entries
    assertNotNull(entryBatch);
    assertEquals(3, entryBatch.getWalEntries().size());
    assertEquals(position, entryBatch.getLastWalPosition());
    assertEquals(walPath, entryBatch.getLastWalPath());
    assertEquals(3, entryBatch.getNbRowKeys());

    appendToLog("foo");
    entryBatch = reader.take();
    assertEquals(1, entryBatch.getNbEntries());
    assertEquals("foo", getRow(entryBatch.getWalEntries().get(0)));
  }
Tests that if we stop before hitting the end of a stream, we can continue where we left off
using the last position
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java