void addSource(String peerId, boolean init) throws IOException {
    ReplicationPeer peer = replicationPeers.getPeer(peerId);
    if (
      ReplicationUtils.LEGACY_REGION_REPLICATION_ENDPOINT_NAME
        .equals(peer.getPeerConfig().getReplicationEndpointImpl())
    ) {
      // we do not use this endpoint for region replication any more, see HBASE-26233
      LOG.info("Legacy region replication peer found, skip adding: {}", peer.getPeerConfig());
      return;
    }
    ReplicationQueueId queueId = new ReplicationQueueId(server.getServerName(), peerId);
    ReplicationSourceInterface src =
      createSource(new ReplicationQueueData(queueId, ImmutableMap.of()), peer);
    // synchronized on latestPaths to avoid missing the new log
    synchronized (this.latestPaths) {
      this.sources.put(peerId, src);
      Map<String, NavigableSet<String>> walsByGroup = new HashMap<>();
      this.walsById.put(queueId, walsByGroup);
      // Add the latest wal to that source's queue
      if (!latestPaths.isEmpty()) {
        for (Map.Entry<String, Path> walPrefixAndPath : latestPaths.entrySet()) {
          Path walPath = walPrefixAndPath.getValue();
          NavigableSet<String> wals = new TreeSet<>();
          wals.add(walPath.getName());
          walsByGroup.put(walPrefixAndPath.getKey(), wals);
          if (!init) {
            // Abort RS and throw exception to make add peer failed
            // Ideally we'd better use the current file size as offset so we can skip replicating
            // the data before adding replication peer, but the problem is that the file may not end
            // at a valid entry's ending, and the current WAL Reader implementation can not deal
            // with reading from the middle of a WAL entry. Can improve later.
            abortAndThrowIOExceptionWhenFail(
              () -> this.queueStorage.setOffset(queueId, walPrefixAndPath.getKey(),
                new ReplicationGroupOffset(walPath.getName(), 0), Collections.emptyMap()));
          }
          src.enqueueLog(walPath);
          LOG.trace("Enqueued {} to source {} during source creation.", walPath, src.getQueueId());
        }
      }
    }
    ReplicationPeerConfig peerConfig = peer.getPeerConfig();
    if (peerConfig.isSyncReplication()) {
      syncReplicationPeerMappingManager.add(peer.getId(), peerConfig);
    }
    src.startup();
  }
Add a normal source for the given peer on this region server. Meanwhile, add new replication
queue to storage. For the newly added peer, we only need to enqueue the latest log of each wal
group and do replication.
<p/>
We add a {@code init} parameter to indicate whether this is part of the initialization process.
If so, we should skip adding the replication queues as this may introduce dead lock on region
server start up and hbase:replication table online.
@param peerId the id of the replication peer
@param init   whether this call is part of the initialization process
@return the source that was created
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java