public static ReplicationPeerManager create(MasterServices services, String clusterId)
    throws ReplicationException, IOException {
    Configuration conf = services.getConfiguration();
    FileSystem fs = services.getMasterFileSystem().getFileSystem();
    ZKWatcher zk = services.getZooKeeper();
    ReplicationPeerStorage peerStorage =
      ReplicationStorageFactory.getReplicationPeerStorage(fs, zk, conf);
    Pair<ReplicationQueueStorage, ReplicationQueueStorageInitializer> pair =
      createReplicationQueueStorage(services);
    ReplicationQueueStorage queueStorage = pair.getFirst();
    ConcurrentMap<String, ReplicationPeerDescription> peers = new ConcurrentHashMap<>();
    for (String peerId : peerStorage.listPeerIds()) {
      ReplicationPeerConfig peerConfig = peerStorage.getPeerConfig(peerId);
      if (
        ReplicationUtils.LEGACY_REGION_REPLICATION_ENDPOINT_NAME
          .equals(peerConfig.getReplicationEndpointImpl())
      ) {
        // If memstore region replication is enabled, there will be a special replication peer
        // usually called 'region_replica_replication'. We do not need to load it or migrate its
        // replication queue data since we do not rely on general replication framework for
        // region replication in 3.x now, please see HBASE-26233 for more details.
        // We can not delete it now since region server with old version still want to update
        // the replicated wal position to zk, if we delete the replication queue zk node, rs
        // will crash. See HBASE-29169 for more details.
        // In MigrateReplicationQueueFromZkToTableProcedure, finally we will call a deleteAllData on
        // the old replication queue storage, to make sure that we will delete the the queue data
        // for this peer and also the peer info in replication peer storage
        LOG.info("Found old region replica replication peer '{}', skip loading it", peerId);
        continue;
      }
      peerConfig = ReplicationPeerConfigUtil.updateReplicationBasePeerConfigs(conf, peerConfig);
      peerStorage.updatePeerConfig(peerId, peerConfig);
      boolean enabled = peerStorage.isPeerEnabled(peerId);
      SyncReplicationState state = peerStorage.getPeerSyncReplicationState(peerId);
      peers.put(peerId, new ReplicationPeerDescription(peerId, enabled, peerConfig, state));
    }
    return new ReplicationPeerManager(fs, zk, peerStorage, queueStorage, peers, conf, clusterId,
      pair.getSecond());
  }
Set a namespace in the peer config means that all tables in this namespace will be replicated
to the peer cluster.
<ol>
<li>If peer config already has a namespace, then not allow set any table of this namespace to
the peer config.</li>
<li>If peer config already has a table, then not allow set this table's namespace to the peer
config.</li>
</ol>
<p>
Set a exclude namespace in the peer config means that all tables in this namespace can't be
replicated to the peer cluster.
<ol>
<li>If peer config already has a exclude namespace, then not allow set any exclude table of
this namespace to the peer config.</li>
<li>If peer config already has a exclude table, then not allow set this table's namespace as a
exclude namespace.</li>
</ol>
hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java