void clearWALEntryBatch() {
    long timeout = EnvironmentEdgeManager.currentTime() + this.shipEditsTimeout;
    while (this.isAlive() || this.entryReader.isAlive()) {
      try {
        if (EnvironmentEdgeManager.currentTime() >= timeout) {
          LOG.warn(
            "Shipper clearWALEntryBatch method timed out whilst waiting reader/shipper "
              + "thread to stop. Not cleaning buffer usage. Shipper alive: {}; Reader alive: {}",
            this.source.getPeerId(), this.isAlive(), this.entryReader.isAlive());
          return;
        } else {
          // Wait both shipper and reader threads to stop
          Thread.sleep(this.sleepForRetries);
        }
      } catch (InterruptedException e) {
        LOG.warn("{} Interrupted while waiting {} to stop on clearWALEntryBatch. "
          + "Not cleaning buffer usage: {}", this.source.getPeerId(), this.getName(), e);
        return;
      }
    }
    long totalReleasedBytes = 0;
    while (true) {
      WALEntryBatch batch = entryReader.entryBatchQueue.poll();
      if (batch == null) {
        break;
      }
      totalReleasedBytes += source.getSourceManager().releaseWALEntryBatchBufferQuota(batch);
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("Decrementing totalBufferUsed by {}B while stopping Replication WAL Readers.",
        totalReleasedBytes);
    }
  }
Attempts to properly update <code>ReplicationSourceManager.totalBufferUser</code>, in case
there were unprocessed entries batched by the reader to the shipper, but the shipper didn't
manage to ship those because the replication source is being terminated. In that case, it
iterates through the batched entries and decrease the pending entries size from
<code>ReplicationSourceManager.totalBufferUser</code>
<p/>
<b>NOTES</b> 1) This method should only be called upon replication source termination. It
blocks waiting for both shipper and reader threads termination, to make sure no race conditions
when updating <code>ReplicationSourceManager.totalBufferUser</code>. 2) It <b>does not</b>
attempt to terminate reader and shipper threads. Those <b>must</b> have been triggered
interruption/termination prior to calling this method.
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java