public void refreshSources(String peerId) throws ReplicationException, IOException {
    String terminateMessage = "Peer " + peerId
      + " state or config changed. Will close the previous replication source and open a new one";
    ReplicationPeer peer = replicationPeers.getPeer(peerId);
    ReplicationQueueId queueId = new ReplicationQueueId(server.getServerName(), peerId);
    ReplicationSourceInterface src;
    // synchronized on latestPaths to avoid missing the new log
    synchronized (this.latestPaths) {
      ReplicationSourceInterface toRemove = this.sources.remove(peerId);
      if (toRemove != null) {
        LOG.info("Terminate replication source for " + toRemove.getPeerId());
        // Do not clear metrics
        toRemove.terminate(terminateMessage, null, false);
      }
      src = createRefreshedSource(queueId, peer);
      this.sources.put(peerId, src);
      for (NavigableSet<String> walsByGroup : walsById.get(queueId).values()) {
        walsByGroup.forEach(wal -> src.enqueueLog(new Path(this.logDir, wal)));
      }
    }
    LOG.info("Startup replication source for " + src.getPeerId());
    src.startup();

    List<ReplicationSourceInterface> toStartup = new ArrayList<>();
    // synchronized on oldsources to avoid race with NodeFailoverWorker
    synchronized (this.oldsources) {
      List<ReplicationQueueId> oldSourceQueueIds = new ArrayList<>();
      for (Iterator<ReplicationSourceInterface> iter = this.oldsources.iterator(); iter
        .hasNext();) {
        ReplicationSourceInterface oldSource = iter.next();
        if (oldSource.getPeerId().equals(peerId)) {
          oldSourceQueueIds.add(oldSource.getQueueId());
          oldSource.terminate(terminateMessage);
          iter.remove();
        }
      }
      for (ReplicationQueueId oldSourceQueueId : oldSourceQueueIds) {
        ReplicationSourceInterface recoveredReplicationSource =
          createRefreshedSource(oldSourceQueueId, peer);
        this.oldsources.add(recoveredReplicationSource);
        for (NavigableSet<Path> walsByGroup : walsByIdRecoveredQueues.get(oldSourceQueueId)
          .values()) {
          walsByGroup.forEach(wal -> recoveredReplicationSource.enqueueLog(wal));
        }
        toStartup.add(recoveredReplicationSource);
      }
    }
    for (ReplicationSourceInterface replicationSource : toStartup) {
      replicationSource.startup();
    }
  }
Close the previous replication sources of this peer id and open new sources to trigger the new
replication state changes or new replication config changes. Here we don't need to change
replication queue storage and only to enqueue all logs to the new replication source
@param peerId the id of the replication peer
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java