@Override
  public void enqueueLog(Path wal) {
    if (!this.filterInWALs.test(wal)) {
      LOG.trace("NOT replicating {}", wal);
      return;
    }
    // Use WAL prefix as the WALGroupId for this peer.
    String walGroupId = AbstractFSWALProvider.getWALPrefixFromWALName(wal.getName());
    boolean queueExists = logQueue.enqueueLog(wal, walGroupId);

    if (!queueExists) {
      if (this.isSourceActive() && this.walEntryFilter != null) {
        // new wal group observed after source startup, start a new worker thread to track it
        // notice: it's possible that wal enqueued when this.running is set but worker thread
        // still not launched, so it's necessary to check workerThreads before start the worker
        tryStartNewShipper(walGroupId);
      }
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("{} Added wal {} to queue of source {}.", logPeerId(), walGroupId, queueId);
    }
  }
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