private void tryStartNewShipper(String walGroupId) {
    workerThreads.compute(walGroupId, (key, value) -> {
      if (value != null) {
        LOG.debug("{} preempted start of shipping worker walGroupId={}", logPeerId(), walGroupId);
        return value;
      } else {
        LOG.debug("{} starting shipping worker for walGroupId={}", logPeerId(), walGroupId);
        ReplicationSourceShipper worker = createNewShipper(walGroupId);
        startShipper(worker);
        return worker;
      }
    });
  }
Instantiation method used by region servers
@param conf      configuration to use
@param fs        file system to use
@param manager   replication manager to ping to
@param server    the server for this region server
@param queueData the id and offsets of our replication queue
@param clusterId unique UUID for the cluster
@param metrics   metrics for replication source
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java