@Override
    public void onReplicaAddedOrRemoved(List<ReplicaId> addedReplicas, List<ReplicaId> removedReplicas) {
      logger.trace("onReplicaAddedOrRemoved added: {}, removed: {}", addedReplicas, removedReplicas);
      // 1. wait for start() to complete
      try {
        startupLatch.await();
      } catch (InterruptedException e) {
        logger.warn("Waiting for startup is interrupted.");
        throw new IllegalStateException("Replication manager startup is interrupted while updating remote replicas");
      }
      if (started) {
        rwLock.writeLock().lock();
        try {
          // 2. determine if added/removed replicas have peer replica on local node.
          //    We skip the replica on current node because it should already be added/removed by state transition thread.
          Set<ReplicaId> addedPeerReplicas = addedReplicas.stream()
              .filter(r -> partitionToPartitionInfo.containsKey(r.getPartitionId()) && r.getDataNodeId() != dataNodeId)
              .collect(Collectors.toSet());
          Set<ReplicaId> removedPeerReplicas = removedReplicas.stream()
              .filter(r -> partitionToPartitionInfo.containsKey(r.getPartitionId()) && r.getDataNodeId() != dataNodeId)
              .collect(Collectors.toSet());

          // No additional synchronization is required because cluster change handler of each dc only updates replica-threads
          // belonging to certain dc. Hence, there is only one thread adding/removing remote replicas within a certain dc.

          // 3. create replicaInfo for new remote replicas and assign them to replica-threads.
          List<RemoteReplicaInfo> replicaInfosToAdd = new ArrayList<>();
          for (ReplicaId remoteReplica : addedPeerReplicas) {
            logger.info("Attempting to add remote replica in replication manager: " + remoteReplica);
            PartitionInfo partitionInfo = partitionToPartitionInfo.get(remoteReplica.getPartitionId());
            // create findToken, remoteReplicaInfo
            FindToken findToken =
                tokenHelper.getFindTokenFactoryFromReplicaType(remoteReplica.getReplicaType()).getNewFindToken();
            RemoteReplicaInfo remoteReplicaInfo =
                createRemoteReplica(remoteReplica, partitionInfo.getLocalReplicaId(), partitionInfo.getStore(), findToken);
            logger.info("Adding remote replica {} on {} to partition info.", remoteReplica.getReplicaPath(),
                remoteReplica.getDataNodeId());
            if (partitionInfo.addReplicaInfoIfAbsent(remoteReplicaInfo)) {
              replicationMetrics.addMetricsForRemoteReplicaInfo(remoteReplicaInfo,
                  replicationConfig.replicationTrackPerDatacenterLagFromLocal,
                  replicationConfig.replicationTrackPerReplicaReplicationBytes);
              replicaInfosToAdd.add(remoteReplicaInfo);
            }
          }
          addRemoteReplicaInfoToReplicaThread(replicaInfosToAdd, true);

          // 4. remove replicaInfo from existing partitionInfo and replica-threads
          List<RemoteReplicaInfo> replicaInfosToRemove = new ArrayList<>();
          for (ReplicaId remoteReplica : removedPeerReplicas) {
            logger.info("Attempting to remove remote replica from replication manager: " + remoteReplica);
            PartitionInfo partitionInfo = partitionToPartitionInfo.get(remoteReplica.getPartitionId());
            RemoteReplicaInfo removedReplicaInfo = partitionInfo.removeReplicaInfoIfPresent(remoteReplica);
            if (removedReplicaInfo != null) {
              replicationMetrics.removeMetricsForRemoteReplicaInfo(removedReplicaInfo);
              logger.info("Removing remote replica {} on {} from replica threads.", remoteReplica.getReplicaPath(),
                  remoteReplica.getDataNodeId());
              replicaInfosToRemove.add(removedReplicaInfo);
            }
          }
          removeRemoteReplicaInfoFromReplicaThread(replicaInfosToRemove);
          if (addedPeerReplicas.size() + removedPeerReplicas.size() > 0) {
            // print only if something is added or removed
            logger.info("{} peer replicas are added and {} peer replicas are removed in replication manager",
                addedPeerReplicas.size(), removedPeerReplicas.size());
          }
        } finally {
          rwLock.writeLock().unlock();
        }
      }
    }
{@inheritDoc}
Note that, this method should be thread-safe because multiple threads (from different cluster change handlers) may
concurrently update remote replica infos.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicationEngine.java