@Test
  public void testWALEntryStreamEOFRightAfterHeader() throws Exception {
    assertEquals(1, logQueue.getQueueSize(fakeWalGroupId));
    AbstractFSWAL<?> abstractWAL = (AbstractFSWAL<?>) log;
    Path emptyLogFile = abstractWAL.getCurrentFileName();
    log.rollWriter(true);

    // AsyncFSWAl and FSHLog both moves the log from WALs to oldWALs directory asynchronously.
    // Wait for in flight wal close count to become 0. This makes sure that empty wal is moved to
    // oldWALs directory.
    Waiter.waitFor(CONF, 5000,
      (Waiter.Predicate<Exception>) () -> abstractWAL.getInflightWALCloseCount() == 0);
    // There will 2 logs in the queue.
    assertEquals(2, logQueue.getQueueSize(fakeWalGroupId));
    appendToLogAndSync();

    Path archivedEmptyLogFile = AbstractFSWALProvider.findArchivedLog(emptyLogFile, CONF);

    // read the wal header
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    bos.write(AbstractProtobufWALReader.PB_WAL_MAGIC);
    try (FSDataInputStream in = fs.open(archivedEmptyLogFile)) {
      IOUtils.skipFully(in, AbstractProtobufWALReader.PB_WAL_MAGIC.length);
      WALHeader header = WALHeader.parseDelimitedFrom(in);
      header.writeDelimitedTo(bos);
    }
    // truncate the first empty log so we have an incomplete header
    try (FSDataOutputStream out = fs.create(archivedEmptyLogFile, true)) {
      bos.writeTo(out);
    }
    try (WALEntryStream entryStream =
      new WALEntryStream(logQueue, fs, CONF, 0, log, new MetricsSource("1"), fakeWalGroupId)) {
      assertEquals(HasNext.RETRY_IMMEDIATELY, entryStream.hasNext());
      assertNotNull(next(entryStream));
    }
  }
This test is for HBASE-27778, when {@link WALEntryFilter#filter} throws exception for some
entries in {@link WALEntryBatch},{@link ReplicationSourceWALReader#totalBufferUsed} should be
decreased because {@link WALEntryBatch} is not put to
{@link ReplicationSourceWALReader#entryBatchQueue}.
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java