@Test
  public void instanceConfigInvalidInfoEntryTest() throws IOException {
    Properties properties = new Properties();
    properties.putAll(props);
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(new VerifiableProperties(properties));
    HelixClusterManager helixClusterManager =
        new HelixClusterManager(clusterMapConfig, selfInstanceName, helixManagerFactory, new MetricRegistry());

    Counter initFailureCount = new Counter();
    HelixClusterChangeHandler helixClusterChangeHandler =
        helixClusterManager.new HelixClusterChangeHandler(localDc, clusterMapConfig.clusterMapClusterName,
            e -> initFailureCount.inc(), false);
    // create an InstanceConfig with invalid entry that mocks error info added by Helix controller
    PartitionId selectedPartition = testPartitionLayout.getPartitionLayout().getPartitions(null).get(0);
    Replica testReplica = (Replica) selectedPartition.getReplicaIds().get(0);
    DataNode testNode = (DataNode) testReplica.getDataNodeId();
    InstanceConfig instanceConfig = new InstanceConfig(getInstanceName(testNode.getHostname(), testNode.getPort()));
    instanceConfig.setHostName(testNode.getHostname());
    instanceConfig.setPort(Integer.toString(testNode.getPort()));
    instanceConfig.getRecord().setSimpleField(ClusterMapUtils.DATACENTER_STR, testNode.getDatacenterName());
    instanceConfig.getRecord().setSimpleField(ClusterMapUtils.RACKID_STR, testNode.getRackId());
    instanceConfig.getRecord()
        .setSimpleField(ClusterMapUtils.SCHEMA_VERSION_STR, Integer.toString(ClusterMapUtils.CURRENT_SCHEMA_VERSION));
    instanceConfig.getRecord().setListField(ClusterMapUtils.SEALED_STR, Collections.emptyList());
    instanceConfig.getRecord().setListField(PARTIALLY_SEALED_STR, Collections.emptyList());
    instanceConfig.getRecord().setListField(ClusterMapUtils.STOPPED_REPLICAS_STR, Collections.emptyList());

    Map<String, Map<String, String>> diskInfos = new HashMap<>();
    assertNotNull("testReplica should not be null", testReplica);
    Map<String, String> diskInfo = new HashMap<>();
    diskInfo.put(ClusterMapUtils.DISK_CAPACITY_STR, Long.toString(testReplica.getDiskId().getRawCapacityInBytes()));
    diskInfo.put(ClusterMapUtils.DISK_STATE, ClusterMapUtils.AVAILABLE_STR);
    String replicasStrBuilder =
        testReplica.getPartition().getId() + ClusterMapUtils.REPLICAS_STR_SEPARATOR + testReplica.getCapacityInBytes()
            + ClusterMapUtils.REPLICAS_STR_SEPARATOR + testReplica.getPartition().getPartitionClass()
            + ClusterMapUtils.REPLICAS_DELIM_STR;
    diskInfo.put(ClusterMapUtils.REPLICAS_STR, replicasStrBuilder);
    diskInfos.put(testReplica.getDiskId().getMountPath(), diskInfo);
    // add an invalid entry at the end of diskInfos
    Map<String, String> invalidEntry = new HashMap<>();
    invalidEntry.put("INVALID_KEY", "INVALID_VALUE");
    diskInfos.put("INVALID_MOUNT_PATH", invalidEntry);
    instanceConfig.getRecord().setMapFields(diskInfos);
    // we call onInstanceConfigChange() twice
    InstanceConfigToDataNodeConfigAdapter.Converter converter =
        new InstanceConfigToDataNodeConfigAdapter.Converter(clusterMapConfig);
    // 1st call, to verify initialization code path
    helixClusterChangeHandler.onDataNodeConfigChange(Collections.singleton(converter.convert(instanceConfig)));
    // 2nd call, to verify dynamic update code path
    helixClusterChangeHandler.onDataNodeConfigChange(Collections.singletonList(converter.convert(instanceConfig)));
    assertEquals("There shouldn't be initialization errors", 0, initFailureCount.getCount());
    helixClusterManager.close();
  }
Test that {@link HelixClusterChangeHandler} is able to handle invalid info entry in the InstanceConfig at runtime
or during initialization.
ambry-clustermap/src/test/java/com/github/ambry/clustermap/ClusterChangeHandlerTest.java