@Test
  public void testReplicationSourceWALReaderDisabled()
    throws IOException, InterruptedException, ExecutionException {
    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();
    ReplicationSource source = mockReplicationSource(false, CONF);
    AtomicInteger invokeCount = new AtomicInteger(0);
    AtomicBoolean enabled = new AtomicBoolean(false);
    when(source.isPeerEnabled()).then(i -> {
      invokeCount.incrementAndGet();
      return enabled.get();
    });

    ReplicationSourceWALReader reader = new ReplicationSourceWALReader(fs, CONF, logQueue, 0,
      getDummyFilter(), source, fakeWalGroupId);
    reader.start();
    Future<WALEntryBatch> future = ForkJoinPool.commonPool().submit(() -> {
      return reader.take();
    });
    // make sure that the isPeerEnabled has been called several times
    TEST_UTIL.waitFor(30000, () -> invokeCount.get() >= 5);
    // confirm that we can read nothing if the peer is disabled
    assertFalse(future.isDone());
    // then enable the peer, we should get the batch
    enabled.set(true);
    WALEntryBatch entryBatch = future.get();

    // 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());
  }
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