@Override
  public void run() {
    int sleepMultiplier = 1;
    while (isReaderRunning()) { // we only loop back here if something fatal happened to our stream
      try (WALEntryStream entryStream = new WALEntryStream(logQueue, fs, conf, currentPosition,
        source.getWALFileLengthProvider(), source.getSourceMetrics(), walGroupId)) {
        while (isReaderRunning()) { // loop here to keep reusing stream while we can
          if (!source.isPeerEnabled()) {
            Threads.sleep(sleepForRetries);
            continue;
          }
          if (!checkBufferQuota()) {
            continue;
          }
          Path currentPath = entryStream.getCurrentPath();
          WALEntryStream.HasNext hasNext = entryStream.hasNext();
          if (hasNext == WALEntryStream.HasNext.NO) {
            replicationDone();
            return;
          }
          // first, check if we have switched a file, if so, we need to manually add an EOF entry
          // batch to the queue
          if (currentPath != null && switched(entryStream, currentPath)) {
            entryBatchQueue.put(WALEntryBatch.endOfFile(currentPath));
            continue;
          }
          if (hasNext == WALEntryStream.HasNext.RETRY) {
            // sleep and retry
            sleepMultiplier = sleep(sleepMultiplier);
            continue;
          }
          if (hasNext == WALEntryStream.HasNext.RETRY_IMMEDIATELY) {
            // retry immediately, this usually means we have switched a file
            continue;
          }
          // below are all for hasNext == YES
          WALEntryBatch batch = createBatch(entryStream);
          boolean successAddToQueue = false;
          try {
            readWALEntries(entryStream, batch);
            currentPosition = entryStream.getPosition();
            // need to propagate the batch even it has no entries since it may carry the last
            // sequence id information for serial replication.
            LOG.debug("Read {} WAL entries eligible for replication", batch.getNbEntries());
            entryBatchQueue.put(batch);
            successAddToQueue = true;
            sleepMultiplier = 1;
          } finally {
            if (!successAddToQueue) {
              // batch is not put to ReplicationSourceWALReader#entryBatchQueue,so we should
              // decrease ReplicationSourceWALReader.totalBufferUsed by the byte size which
              // acquired in ReplicationSourceWALReader.acquireBufferQuota.
              this.getSourceManager().releaseWALEntryBatchBufferQuota(batch);
            }
          }
        }
      } catch (WALEntryFilterRetryableException e) {
        // here we have to recreate the WALEntryStream, as when filtering, we have already called
        // next to get the WAL entry and advanced the WALEntryStream, at WALEntryStream layer, it
        // just considers everything is fine,that's why the catch block is not in the inner block
        LOG.warn("Failed to filter WAL entries and the filter let us retry later", e);
        sleepMultiplier = sleep(sleepMultiplier);
      } catch (InterruptedException e) {
        // this usually means we want to quit
        LOG.warn("Interrupted while sleeping between WAL reads or adding WAL batch to ship queue",
          e);
        Thread.currentThread().interrupt();
      }
    }
  }
Creates a reader worker for a given WAL queue. Reads WAL entries off a given queue, batches the
entries, and puts them on a batch queue.
@param fs            the files system to use
@param conf          configuration to use
@param logQueue      The WAL queue to read off of
@param startPosition position in the first WAL to start reading from
@param filter        The filter to use while reading
@param source        replication source
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java