@Override
  public final void run() {
    setWorkerState(WorkerState.RUNNING);
    LOG.info("Running ReplicationSourceShipper Thread for wal group: {}", this.walGroupId);
    // Loop until we close down
    while (isActive()) {
      // Sleep until replication is enabled again
      if (!source.isPeerEnabled()) {
        // The peer enabled check is in memory, not expensive, so do not need to increase the
        // sleep interval as it may cause a long lag when we enable the peer.
        sleepForRetries("Replication is disabled", sleepForRetries, 1, maxRetriesMultiplier);
        continue;
      }
      try {
        WALEntryBatch entryBatch = entryReader.poll(getEntriesTimeout);
        LOG.debug("Shipper from source {} got entry batch from reader: {}", source.getQueueId(),
          entryBatch);
        if (entryBatch == null) {
          continue;
        }
        // the NO_MORE_DATA instance has no path so do not call shipEdits
        if (entryBatch == WALEntryBatch.NO_MORE_DATA) {
          noMoreData();
        } else {
          shipEdits(entryBatch);
        }
      } catch (InterruptedException | ReplicationRuntimeException e) {
        // It is interrupted and needs to quit.
        LOG.warn("Interrupted while waiting for next replication entry batch", e);
        Thread.currentThread().interrupt();
      }
    }
    // If the worker exits run loop without finishing its task, mark it as stopped.
    if (!isFinished()) {
      setWorkerState(WorkerState.STOPPED);
    } else {
      source.removeWorker(this);
      postFinish();
    }
  }
This thread reads entries from a queue and ships them. Entries are placed onto the queue by
ReplicationSourceWALReaderThread
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java