@Test
  public void addNewInstancesAndPartitionsTest() 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());
    // before adding new instances, let's first ensure current number of nodes is correct.
    List<DataNode> dataNodesInLayout = new ArrayList<>();
    testHardwareLayout.getHardwareLayout().getDatacenters().forEach(dc -> dataNodesInLayout.addAll(dc.getDataNodes()));
    assertEquals("Number of data nodes is not expected", dataNodesInLayout.size(),
        helixClusterManager.getDataNodeIds().size());

    // pick 2 existing nodes from each dc (also the nodes from local dc should be different from currentNode)
    List<DataNode> nodesToHostNewPartition = new ArrayList<>();
    List<DataNode> localDcNodes = testHardwareLayout.getAllDataNodesFromDc(localDc);
    localDcNodes.remove(currentNode);
    DataNode localDcNode1 = localDcNodes.get(0);
    DataNode localDcNode2 = localDcNodes.get(localDcNodes.size() - 1);
    List<DataNode> remoteDcNodes = testHardwareLayout.getAllDataNodesFromDc(remoteDc);
    DataNode remoteDcNode1 = remoteDcNodes.get(0);
    DataNode remoteDcNode2 = remoteDcNodes.get(remoteDcNodes.size() - 1);

    // add a new node into static layout
    testHardwareLayout.addNewDataNodes(1);
    // add a new partition to static layout and put its replicas to both existing nodes and new node
    List<DataNode> newAddedNodes = new ArrayList<>();
    testHardwareLayout.getHardwareLayout().getDatacenters().forEach(dc -> newAddedNodes.addAll(dc.getDataNodes()));
    newAddedNodes.removeAll(dataNodesInLayout);
    // pick 2 existing nodes and 1 new node from each dc to add a replica from new partition
    nodesToHostNewPartition.addAll(Arrays.asList(localDcNode1, localDcNode2));
    nodesToHostNewPartition.addAll(Arrays.asList(remoteDcNode1, remoteDcNode2));
    nodesToHostNewPartition.addAll(newAddedNodes);
    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.
    // In each dc, 2 existing instance configs are updated and 1 new instance is added as well as 1 new partition
    helixCluster.upgradeWithNewHardwareLayout(hardwareLayoutPath);

    // verify after InstanceConfig change, HelixClusterManager contains the one more node per dc.
    assertEquals("Number of data nodes after instance addition is not correct",
        testHardwareLayout.getAllExistingDataNodes().size(), helixClusterManager.getDataNodeIds().size());
    // verify number of partitions in cluster manager has increased by 1
    assertEquals("Number of partitions after partition addition is not correct",
        testPartitionLayout.getPartitionCount(), helixClusterManager.getAllPartitionIds(null).size());
    // verify writable partitions in HelixClusterManager with partition override enabled/disabled (to call getWritablePartitionIds,
    // we need to bring new added instances up because that method checks if all local replicas are up)
    for (DataNode newNode : newAddedNodes) {
      helixCluster.bringInstanceUp(getInstanceName(newNode.getHostname(), newNode.getPort()));
    }
    assertEquals("Number of writable partitions after partition addition is not correct",
        overrideEnabled ? testPartitionLayout.getPartitionCount() - 1 : testPartitionLayout.getPartitionCount(),
        helixClusterManager.getWritablePartitionIds(null).size());

    // verify capacity stats are updated
    HelixClusterManagerQueryHelper clusterManagerCallback = helixClusterManager.getManagerQueryHelper();
    // note that we add one node to each dc, so the raw capacity = (# of nodes) * (# of disks) * (disk capacity)
    long rawCapacityInStaticLayout =
        testHardwareLayout.getAllExistingDataNodes().size() * testHardwareLayout.getDiskCount()
            * testHardwareLayout.getDiskCapacityInBytes();
    assertEquals("Raw capacity of entire cluster is not expected", rawCapacityInStaticLayout,
        clusterManagerCallback.getRawCapacity());
    // we have added one more partition, so now the allocated raw capacity in cluster is 4 (partition count) * 6 * ReplicaCapacity
    assertEquals("Allocated raw capacity of entire cluster is not correct",
        testPartitionLayout.getAllocatedRawCapacityInBytes(), clusterManagerCallback.getAllocatedRawCapacity());
    // verify usable capacity
    assertEquals("Allocated usable capacity of entire cluster is not correct",
        testPartitionLayout.getAllocatedUsableCapacityInBytes(), clusterManagerCallback.getAllocatedUsableCapacity());

    // additional tests to verify getting replicas, disks and resources etc returns correct results.
    for (DataNode newNode : newAddedNodes) {
      AmbryDataNode ambryNode = helixClusterManager.getDataNodeId(newNode.getHostname(), newNode.getPort());
      assertNotNull("New added node should exist in HelixClusterManager", ambryNode);
      List<AmbryReplica> ambryReplicas = helixClusterManager.getReplicaIds(ambryNode);
      assertEquals("There should be one replica on the new node", 1, ambryReplicas.size());
      Set<AmbryDisk> ambryDisks = new HashSet<>(clusterManagerCallback.getDisks(ambryNode));
      assertEquals("Disk count on the new node is not correct", localDcNode1.getDisks().size(), ambryDisks.size());
      // verify that get a non-existent partition on new node should return null
      assertNull("Should return null when getting a non-existent replica on new node",
          helixClusterManager.getReplicaForPartitionOnNode(ambryNode, "0"));
    }
    // trigger IdealState change and refresh partition-to-resource mapping (bring in the new partition in resource map)
    helixCluster.refreshIdealState();
    Map<String, Set<String>> partitionNameToResource = helixClusterManager.getPartitionToResourceMapByDC().get(localDc);
    List<PartitionId> partitionIds = testPartitionLayout.getPartitionLayout().getPartitions(null);
    // verify all partitions (including the new added one) are present in partition-to-resource map
    Set<String> partitionNames = partitionIds.stream().map(PartitionId::toPathString).collect(Collectors.toSet());
    assertEquals("Some partitions are not present in partition-to-resource map", partitionNames,
        partitionNameToResource.keySet());
    // verify all partitions are able to get their resource name
    helixClusterManager.getAllPartitionIds(DEFAULT_PARTITION_CLASS)
        .forEach(partitionId -> assertEquals("Resource name is not expected",
            partitionNameToResource.get(partitionId.toPathString()), new HashSet<>(partitionId.getResourceNames())));
    helixClusterManager.close();
  }
Test new instances/partitions are added to cluster dynamically. {@link HelixClusterManager} with
{@link HelixClusterChangeHandler} should absorb the change and update in-mem cluster map.
1. add new instance
2. add new partition onto new instance
3. add new partition onto existing instance
ambry-clustermap/src/test/java/com/github/ambry/clustermap/ClusterChangeHandlerTest.java