@Test
  public void onRoutingTableUpdateCallbackTest() throws Exception {
    Pair<StorageManager, ReplicationManager> managers =
        createStorageManagerAndReplicationManager(clusterMap, clusterMapConfig, mockHelixParticipant);
    StorageManager storageManager = managers.getFirst();
    MockReplicationManager replicationManager = (MockReplicationManager) managers.getSecond();
    replicationManager.start();

    //Trigger PartitionStateChangeListener callback to replication manager to notify that a local replica state has changed from STANDBY to LEADER
    List<? extends ReplicaId> replicaIds = clusterMap.getReplicaIds(replicationManager.dataNodeId);
    for (ReplicaId replicaId : replicaIds) {
      MockReplicaId mockReplicaId = (MockReplicaId) replicaId;
      if (mockReplicaId.getReplicaState() == ReplicaState.LEADER) {
        MockPartitionId existingPartition = (MockPartitionId) mockReplicaId.getPartitionId();
        mockHelixParticipant.onPartitionBecomeLeaderFromStandby(existingPartition.toPathString());

        //verify that map of peerLeaderReplicasByPartition in PartitionLeaderInfo is updated correctly
        Set<ReplicaId> peerLeaderReplicasInPartitionLeaderInfo =
            replicationManager.leaderBasedReplicationAdmin.getLeaderPartitionToPeerLeaderReplicas()
                .get(existingPartition.toPathString());
        Set<ReplicaId> peerLeaderReplicasInClusterMap =
            new HashSet<>(existingPartition.getReplicaIdsByState(ReplicaState.LEADER, null));
        peerLeaderReplicasInClusterMap.remove(mockReplicaId);
        assertThat(
            "Mismatch in list of leader peer replicas stored by partition in replication manager with cluster map",
            peerLeaderReplicasInPartitionLeaderInfo, is(peerLeaderReplicasInClusterMap));

        //Switch the LEADER/STANDBY states for remote replicas on one of the remote data centers
        ReplicaId peerLeaderReplica = peerLeaderReplicasInClusterMap.iterator().next();
        ReplicaId peerStandByReplica = existingPartition.getReplicaIdsByState(ReplicaState.STANDBY,
            peerLeaderReplica.getDataNodeId().getDatacenterName()).get(0);
        existingPartition.setReplicaState(peerLeaderReplica, ReplicaState.STANDBY);
        existingPartition.setReplicaState(peerStandByReplica, ReplicaState.LEADER);

        //Trigger routing table change callback to replication manager
        ClusterMapChangeListener clusterMapChangeListener = clusterMap.getClusterMapChangeListener();
        clusterMapChangeListener.onRoutingTableChange();

        //verify that new remote leader is reflected in the peerLeaderReplicasByPartition map
        peerLeaderReplicasInPartitionLeaderInfo =
            replicationManager.leaderBasedReplicationAdmin.getLeaderPartitionToPeerLeaderReplicas()
                .get(existingPartition.toPathString());
        peerLeaderReplicasInClusterMap =
            new HashSet<>(existingPartition.getReplicaIdsByState(ReplicaState.LEADER, null));
        peerLeaderReplicasInClusterMap.remove(mockReplicaId);
        assertThat(
            "Mismatch in map of peer leader replicas stored by partition in replication manager with cluster map after routing table update",
            peerLeaderReplicasInPartitionLeaderInfo, is(peerLeaderReplicasInClusterMap));
      }
    }

    storageManager.shutdown();
  }
Test cluster map change callback in {@link ReplicationManager} for routing table updates.
Test setup: When creating partitions, have one replica in LEADER state and rest in STANDBY states on each data center and
later switch the states of replicas (LEADER to STANDBY and STANDBY to LEADER) on one of the DCs during the test
Test condition: When replication manager receives onRoutingTableUpdate() indication after the remote replica states were updated,
map of partition to peer leader replicas stored in replication manager should be updated correctly
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/LeaderBasedReplicationTest.java