private void addPeerAndWait(String peerId, String clusterKey, boolean syncRep)
    throws ReplicationException, IOException {
    ReplicationPeerConfigBuilder builder = ReplicationPeerConfig.newBuilder()
      .setClusterKey(UTIL.getZkCluster().getAddress().toString() + ":/" + clusterKey)
      .setReplicationEndpointImpl(ReplicationEndpointForTest.class.getName());
    if (syncRep) {
      builder.setTableCFsMap(ImmutableMap.of(TABLE_NAME, Collections.emptyList()))
        .setRemoteWALDir(FS.makeQualified(remoteLogDir).toString());
    }

    manager.getReplicationPeers().getPeerStorage().addPeer(peerId, builder.build(), true,
      syncRep ? SyncReplicationState.DOWNGRADE_ACTIVE : SyncReplicationState.NONE);
    manager.addPeer(peerId);
    UTIL.waitFor(20000, () -> {
      ReplicationSourceInterface rs = manager.getSource(peerId);
      return rs != null && rs.isSourceActive();
    });
  }
Add a peer and wait for it to initialize
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java