private void verifyDataNodeAndDiskEquivalencyInDc(Datacenter dc, String clusterName,
      PartitionLayout partitionLayout) {
    String dcName = dc.getName();
    // The following properties are immaterial for the tool, but the ClusterMapConfig mandates their presence.
    ClusterMapConfig clusterMapConfig = getClusterMapConfig(clusterName, dcName, null);
    StaticClusterManager staticClusterMap =
        (new StaticClusterAgentsFactory(clusterMapConfig, partitionLayout)).getClusterMap();
    String zkConnectStr = dataCenterToZkAddress.get(dcName).getZkConnectStrs().get(0);
    try (PropertyStoreToDataNodeConfigAdapter propertyStoreAdapter = new PropertyStoreToDataNodeConfigAdapter(
        zkConnectStr, clusterMapConfig)) {
      InstanceConfigToDataNodeConfigAdapter.Converter instanceConfigConverter =
          new InstanceConfigToDataNodeConfigAdapter.Converter(clusterMapConfig);
      Set<String> allInstancesInHelix = new HashSet<>(getInstanceNamesInHelix(dcName, propertyStoreAdapter));
      for (DataNodeId dataNodeId : dc.getDataNodes()) {
        Map<String, Map<String, ReplicaId>> mountPathToReplicas = getMountPathToReplicas(staticClusterMap, dataNodeId);
        DataNode dataNode = (DataNode) dataNodeId;
        String instanceName = getInstanceName(dataNode);
        ensureOrThrow(allInstancesInHelix.remove(instanceName), "Instance not present in Helix " + instanceName);
        DataNodeConfig dataNodeConfig =
            getDataNodeConfigFromHelix(clusterName, instanceName, propertyStoreAdapter, instanceConfigConverter,
                adminForDc.get(dcName), dataNodeConfigSourceType);
        Map<String, DataNodeConfig.DiskConfig> diskInfos = new HashMap<>(dataNodeConfig.getDiskConfigs());
        for (Disk disk : dataNode.getDisks()) {
          DataNodeConfig.DiskConfig diskInfoInHelix = diskInfos.remove(disk.getMountPath());
          ensureOrThrow(diskInfoInHelix != null,
              "[" + dcName.toUpperCase() + "] Disk not present for instance " + instanceName + " disk "
                  + disk.getMountPath());
          ensureOrThrow(disk.getRawCapacityInBytes() == diskInfoInHelix.getDiskCapacityInBytes(),
              "[" + dcName.toUpperCase() + "] Capacity mismatch for instance " + instanceName + " disk "
                  + disk.getMountPath());

          // We check replicaInfo only when this is a Bootstrap or Validate operation. For other operations, replica infos
          // are expected to be different on certain nodes.
          if (EnumSet.of(HelixAdminOperation.BootstrapCluster, HelixAdminOperation.ValidateCluster)
              .contains(helixAdminOperation)) {
            Set<String> replicasInClusterMap = new HashSet<>();
            Map<String, ReplicaId> replicaList = mountPathToReplicas.get(disk.getMountPath());
            if (replicaList != null) {
              replicasInClusterMap.addAll(replicaList.keySet());
            }
            Set<String> replicasInHelix = new HashSet<>();
            Map<String, DataNodeConfig.ReplicaConfig> replicaConfigMap = diskInfoInHelix.getReplicaConfigs();
            for (Map.Entry<String, DataNodeConfig.ReplicaConfig> replicaConfigEntry : replicaConfigMap.entrySet()) {
              String replicaName = replicaConfigEntry.getKey();
              DataNodeConfig.ReplicaConfig replicaConfig = replicaConfigEntry.getValue();
              replicasInHelix.add(replicaName);
              ReplicaId replica = replicaList.get(replicaName);
              ensureOrThrow(replicaConfig.getReplicaCapacityInBytes() == replica.getCapacityInBytes(),
                  "[" + dcName.toUpperCase() + "] Replica capacity should be the same.");
              ensureOrThrow(replicaConfig.getPartitionClass().equals(replica.getPartitionId().getPartitionClass()),
                  "[" + dcName.toUpperCase() + "] Partition class should be the same.");
            }
            ensureOrThrow(replicasInClusterMap.equals(replicasInHelix),
                "[" + dcName.toUpperCase() + "] Replica information not consistent for instance " + instanceName
                    + " disk " + disk.getMountPath() + "\n in Helix: " + replicasInHelix + "\n in static clustermap: "
                    + replicasInClusterMap);
          }
        }
        for (Map.Entry<String, DataNodeConfig.DiskConfig> entry : diskInfos.entrySet()) {
          String mountPath = entry.getKey();
          if (!mountPath.startsWith("/mnt")) {
            logger.warn("[{}] Instance {} has unidentifiable mount path in Helix: {}", dcName.toUpperCase(),
                instanceName, mountPath);
          } else {
            throw new AssertionError(
                "[" + dcName.toUpperCase() + "] Instance " + instanceName + " has extra disk in Helix: " + entry);
          }
        }
        ensureOrThrow(!dataNode.hasSSLPort() || (dataNode.getSSLPort() == dataNodeConfig.getSslPort()),
            "[" + dcName.toUpperCase() + "] SSL Port mismatch for instance " + instanceName);
        ensureOrThrow(!dataNode.hasHttp2Port() || (dataNode.getHttp2Port() == dataNodeConfig.getHttp2Port()),
            "[" + dcName.toUpperCase() + "] HTTP2 Port mismatch for instance " + instanceName);
        ensureOrThrow(dataNode.getDatacenterName().equals(dataNodeConfig.getDatacenterName()),
            "[" + dcName.toUpperCase() + "] Datacenter mismatch for instance " + instanceName);
        ensureOrThrow(Objects.equals(dataNode.getRackId(), dataNodeConfig.getRackId()),
            "[" + dcName.toUpperCase() + "] Rack Id mismatch for instance " + instanceName);
        // xid is not set in PropertyStore based DataNodeConfig and will be decommissioned eventually, hence we don't check xid equivalence
      }
      if (expectMoreInHelixDuringValidate) {
        ensureOrThrow(
            allInstancesInHelix.equals(instancesNotForceRemovedByDc.getOrDefault(dc.getName(), new HashSet<>())),
            "[" + dcName.toUpperCase() + "] Additional instances in Helix: " + allInstancesInHelix
                + " not what is expected " + instancesNotForceRemovedByDc.get(dc.getName()));
        info("[{}] *** Helix may have more instances than in the given clustermap as removals were not forced.",
            dcName.toUpperCase());
      } else {
        ensureOrThrow(allInstancesInHelix.isEmpty(),
            "[" + dcName.toUpperCase() + "] Following instances in Helix not found in the clustermap "
                + allInstancesInHelix);
      }
    }
    info("[{}] Successfully verified datanode and disk equivalency in dc {}", dcName.toUpperCase(), dc.getName());
  }
Verify that the hardware layout information is in sync - which includes the node and disk information. Also verify
that the replicas belonging to disks are in sync between the static cluster map and Helix.
@param dc the datacenter whose information is to be verified.
@param clusterName the cluster to be verified.
@param partitionLayout the {@link PartitionLayout} of the static clustermap.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/HelixBootstrapUpgradeUtil.java