@Test
  public void testReplicationSourceWALReaderWithPartialWALEntryFailingFilter() 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();
    }

    Path walPath = getQueue().peek();
    int maxThrowExceptionCount = 3;

    ReplicationSource source = mockReplicationSource(false, CONF);
    when(source.isPeerEnabled()).thenReturn(true);
    PartialWALEntryFailingWALEntryFilter walEntryFilter =
      new PartialWALEntryFailingWALEntryFilter(maxThrowExceptionCount, 3);
    ReplicationSourceWALReader reader =
      new ReplicationSourceWALReader(fs, CONF, logQueue, 0, walEntryFilter, source, fakeWalGroupId);
    reader.start();
    WALEntryBatch entryBatch = reader.take();

    assertNotNull(entryBatch);
    assertEquals(3, entryBatch.getWalEntries().size());
    long sum = entryBatch.getWalEntries().stream()
      .mapToLong(WALEntryBatch::getEntrySizeExcludeBulkLoad).sum();
    assertEquals(position, entryBatch.getLastWalPosition());
    assertEquals(walPath, entryBatch.getLastWalPath());
    assertEquals(3, entryBatch.getNbRowKeys());
    assertEquals(sum, source.getSourceManager().getTotalBufferUsed());
    assertEquals(sum, source.getSourceManager().getGlobalMetrics().getWALReaderEditsBufferBytes());
    assertEquals(maxThrowExceptionCount, walEntryFilter.getThrowExceptionCount());
    assertNull(reader.poll(10));
  }
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