@Test
  public void testEOFExceptionForRecoveredQueueWithMultipleLogs() throws Exception {
    Configuration conf = new Configuration(CONF);
    MetricsSource metrics = mock(MetricsSource.class);
    ReplicationSource source = mockReplicationSource(true, conf);
    ReplicationSourceLogQueue localLogQueue = new ReplicationSourceLogQueue(conf, metrics, source);
    // Create a 0 length log.
    Path emptyLog = new Path(fs.getHomeDirectory(), "log.2." + isCompressionEnabled);
    fs.create(emptyLog).close();
    assertEquals(0, fs.getFileStatus(emptyLog).getLen());
    localLogQueue.enqueueLog(emptyLog, fakeWalGroupId);

    final Path log1 = new Path(fs.getHomeDirectory(), "log.1." + isCompressionEnabled);
    WALProvider.Writer writer1 = WALFactory.createWALWriter(fs, log1, TEST_UTIL.getConfiguration());
    appendEntries(writer1, 3);
    localLogQueue.enqueueLog(log1, fakeWalGroupId);

    when(source.isPeerEnabled()).thenReturn(true);
    // Override the max retries multiplier to fail fast.
    conf.setInt("replication.source.maxretriesmultiplier", 1);
    conf.setBoolean("replication.source.eof.autorecovery", true);
    conf.setInt("replication.source.nb.batches", 10);
    // Create a reader thread.
    ReplicationSourceWALReader reader = new ReplicationSourceWALReader(fs, conf, localLogQueue, 0,
      getDummyFilter(), source, fakeWalGroupId);
    assertEquals("Initial log queue size is not correct", 2,
      localLogQueue.getQueueSize(fakeWalGroupId));
    reader.start();
    reader.join();

    // remove empty log from logQueue.
    assertEquals(0, localLogQueue.getQueueSize(fakeWalGroupId));
    assertEquals("Log queue should be empty", 0, localLogQueue.getQueueSize(fakeWalGroupId));
  }
Test removal of 0 length log from logQueue if the source is a recovered source and size of
logQueue is only 1.
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java