@Test
  public void replicaFromStandbyToLeaderTest() throws Exception {
    MockClusterMap clusterMap = new MockClusterMap();
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(verifiableProperties);
    MockHelixParticipant.metricRegistry = new MetricRegistry();
    MockHelixParticipant mockHelixParticipant = new MockHelixParticipant(clusterMapConfig);

    ReplicationConfig initialReplicationConfig = replicationConfig;
    properties.setProperty("replication.model.across.datacenters", "LEADER_BASED");
    replicationConfig = new ReplicationConfig(new VerifiableProperties(properties));

    Pair<StorageManager, ReplicationManager> managers =
        createStorageManagerAndReplicationManager(clusterMap, clusterMapConfig, mockHelixParticipant);
    StorageManager storageManager = managers.getFirst();
    MockReplicationManager replicationManager = (MockReplicationManager) managers.getSecond();

    List<ReplicaId> replicaIds = clusterMap.getReplicaIds(replicationManager.dataNodeId);
    for (ReplicaId replicaId : replicaIds) {
      MockReplicaId mockReplicaId = (MockReplicaId) replicaId;
      if (mockReplicaId.getReplicaState() == ReplicaState.LEADER) {
        PartitionId existingPartition = mockReplicaId.getPartitionId();
        mockHelixParticipant.onPartitionBecomeLeaderFromStandby(existingPartition.toPathString());
        Set<ReplicaId> peerLeaderReplicasInReplicationManager =
            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 and cluster map",
            peerLeaderReplicasInReplicationManager, is(peerLeaderReplicasInClusterMap));
      }
    }
    storageManager.shutdown();

    replicationConfig = initialReplicationConfig;
  }
Test state transition in replication manager from STANDBY to LEADER
Test setup: When creating partitions, make sure that there is exactly one replica in LEADER STATE on each data center
Test condition: When a partition on current node moves from standby to leader, verify that in-memory map storing
partition to peer leader replicas is updated correctly
@throws Exception
ambry-replication/src/test/java/com/github/ambry/replication/ReplicationTest.java