@Test
  public void replicaAdditionOnCurrentNodeTest() throws Exception {
    // create a HelixClusterManager with DynamicClusterChangeHandler
    Properties properties = new Properties();
    properties.putAll(props);
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(new VerifiableProperties(properties));
    HelixClusterManager helixClusterManager =
        new HelixClusterManager(clusterMapConfig, selfInstanceName, helixManagerFactory, new MetricRegistry());
    // test setup: create 2 new partitions and place their replicas onto nodes that exclude currentNode. This is to avoid
    // edge case where currentNode already has all partitions in cluster
    Set<PartitionId> initialPartitionSet = new HashSet<>(testPartitionLayout.getPartitionLayout().getPartitions(null));
    List<DataNode> nodesToHostNewPartition = new ArrayList<>();
    List<DataNode> localDcNodes = testHardwareLayout.getAllDataNodesFromDc(localDc)
        .stream()
        .filter(node -> node != currentNode)
        .collect(Collectors.toList());
    List<DataNode> remoteDcNodes = testHardwareLayout.getAllDataNodesFromDc(remoteDc);
    nodesToHostNewPartition.addAll(localDcNodes.subList(0, 3));
    nodesToHostNewPartition.addAll(remoteDcNodes.subList(0, 3));
    testPartitionLayout.addNewPartition(testHardwareLayout, nodesToHostNewPartition, DEFAULT_PARTITION_CLASS);
    // add one more new partition
    testPartitionLayout.addNewPartition(testHardwareLayout, nodesToHostNewPartition, DEFAULT_PARTITION_CLASS);
    // write new HardwareLayout and PartitionLayout into files
    Utils.writeJsonObjectToFile(testHardwareLayout.getHardwareLayout().toJSONObject(), hardwareLayoutPath);
    Utils.writeJsonObjectToFile(testPartitionLayout.getPartitionLayout().toJSONObject(), partitionLayoutPath);
    // this triggers a InstanceConfig change notification.
    helixCluster.upgradeWithNewHardwareLayout(hardwareLayoutPath);
    Set<PartitionId> updatedPartitionSet = new HashSet<>(testPartitionLayout.getPartitionLayout().getPartitions(null));
    updatedPartitionSet.removeAll(initialPartitionSet);
    List<PartitionId> addedPartitions = new ArrayList<>(updatedPartitionSet);
    assertEquals("There should be 2 added partitions", 2, addedPartitions.size());
    Partition addedPartition1 = (Partition) addedPartitions.get(0);
    Partition addedPartition2 = (Partition) addedPartitions.get(1);
    // add one replica of this newly added partition1 to currentNode
    testPartitionLayout.addReplicaToPartition(currentNode, addedPartition1);
    // before upgrading Helix, let's save the replica count of test partition to a variable
    PartitionId partitionInManager = helixClusterManager.getAllPartitionIds(null)
        .stream()
        .filter(p -> p.toPathString().equals(addedPartition1.toPathString()))
        .findFirst()
        .get();
    int previousReplicaCnt = partitionInManager.getReplicaIds().size();
    // test case 1: without populating bootstrap replica, new replica in InstanceConfig will trigger exception on current
    // node (this shouldn't happen in practice but we still mock this situation to perform exhaustive testing)
    Utils.writeJsonObjectToFile(testPartitionLayout.getPartitionLayout().toJSONObject(), partitionLayoutPath);
    helixCluster.upgradeWithNewPartitionLayout(partitionLayoutPath,
        HelixBootstrapUpgradeUtil.HelixAdminOperation.BootstrapCluster);
    assertEquals("Replica count of testing partition shouldn't change", previousReplicaCnt,
        partitionInManager.getReplicaIds().size());
    // verify there is an exception when handling instance config change due to replica not found in bootstrap replica map
    assertEquals("Instance config change error count should be 1", 1,
        helixClusterManager.helixClusterManagerMetrics.instanceConfigChangeErrorCount.getCount());
    helixClusterManager.close();
    // test case 2: call getBootstrapReplica in HelixClusterManager to populate bootstrap replica map and then upgrade
    // Helix again.
    Map<String, Map<String, String>> partitionToReplicaInfosMap = new HashMap<>();
    Map<String, String> newReplicaInfos = new HashMap<>();
    newReplicaInfos.put(PARTITION_CLASS_STR, DEFAULT_PARTITION_CLASS);
    newReplicaInfos.put(REPLICAS_CAPACITY_STR, String.valueOf(TestPartitionLayout.defaultReplicaCapacityInBytes));
    newReplicaInfos.put(currentNode.getHostname() + "_" + currentNode.getPort(),
        currentNode.getDisks().get(0).getMountPath());
    partitionToReplicaInfosMap.put(addedPartition2.toPathString(), newReplicaInfos);
    // set ZNRecord
    ZNRecord replicaInfosZNRecord = new ZNRecord(REPLICA_ADDITION_STR);
    replicaInfosZNRecord.setMapFields(partitionToReplicaInfosMap);
    znRecordMap.put(REPLICA_ADDITION_ZNODE_PATH, replicaInfosZNRecord);
    // create a new HelixClusterManager with replica addition info in Helix
    helixClusterManager = new HelixClusterManager(clusterMapConfig, selfInstanceName,
        new HelixClusterManagerTest.MockHelixManagerFactory(helixCluster, znRecordMap, null), new MetricRegistry());
    ReplicaId bootstrapReplica = helixClusterManager.getBootstrapReplica(addedPartition2.toPathString(), currentNode);
    assertNotNull("Getting bootstrap replica should succeed", bootstrapReplica);
    // add replica of new partition2 to currentNode
    testPartitionLayout.addReplicaToPartition(currentNode, addedPartition2);
    Utils.writeJsonObjectToFile(testPartitionLayout.getPartitionLayout().toJSONObject(), partitionLayoutPath);
    helixCluster.upgradeWithNewPartitionLayout(partitionLayoutPath,
        HelixBootstrapUpgradeUtil.HelixAdminOperation.BootstrapCluster);

    partitionInManager = helixClusterManager.getAllPartitionIds(null)
        .stream()
        .filter(p -> p.toPathString().equals(addedPartition2.toPathString()))
        .findFirst()
        .get();
    // this time the new replica should be present in bootstrap replica map and therefore replica count should increase
    assertEquals("Replica count of testing partition shouldn't change", previousReplicaCnt + 1,
        partitionInManager.getReplicaIds().size());
    // verify that the replica instance in HelixClusterManager is same with bootstrap replica instance
    ReplicaId replicaInManager = helixClusterManager.getReplicaIds(
            helixClusterManager.getDataNodeId(currentNode.getHostname(), currentNode.getPort()))
        .stream()
        .filter(r -> r.getPartitionId().toPathString().equals(addedPartition2.toPathString()))
        .findFirst()
        .get();
    assertSame("There should be exactly one instance for added replica", replicaInManager, bootstrapReplica);
    helixClusterManager.close();
  }
Test the case where current node receives InstanceConfig change triggered by itself due to replica addition. We need
to verify {@link HelixClusterChangeHandler} will check if new replica from InstanceConfig exists in bootstrap
replica map. The intention here is to avoid creating a second instance of replica on current node.
ambry-clustermap/src/test/java/com/github/ambry/clustermap/ClusterChangeHandlerTest.java