DataNodeConfig convert(InstanceConfig instanceConfig) {
      int schemaVersion = getSchemaVersion(instanceConfig);
      if (schemaVersion != 0) {
        LOGGER.warn("Unknown InstanceConfig schema version {} in {}. Ignoring.", schemaVersion, instanceConfig);
        return null;
      }
      DataNodeConfig dataNodeConfig = new DataNodeConfig(instanceConfig.getInstanceName(), instanceConfig.getHostName(),
          Integer.parseInt(instanceConfig.getPort()), getDcName(instanceConfig), getSslPortStr(instanceConfig),
          getHttp2PortStr(instanceConfig), getRackId(instanceConfig), getXid(instanceConfig));
      dataNodeConfig.getSealedReplicas().addAll(getSealedReplicas(instanceConfig));
      dataNodeConfig.getPartiallySealedReplicas().addAll(getPartiallySealedReplicas(instanceConfig));
      dataNodeConfig.getStoppedReplicas().addAll(getStoppedReplicas(instanceConfig));
      dataNodeConfig.getDisabledReplicas().addAll(getDisabledReplicas(instanceConfig));
      instanceConfig.getRecord().getMapFields().forEach((mountPath, diskProps) -> {
        if (diskProps.get(DISK_STATE) == null) {
          // Check if this map field actually holds disk properties, since we can't tell from just the field key (the
          // mount path with no special prefix). There may be extra fields when Helix controller adds partitions in ERROR
          // state to InstanceConfig.
          LOGGER.info("{} field does not contain disk info on {}. Storing it in extraMapFields.", mountPath,
              instanceConfig.getInstanceName());
          dataNodeConfig.getExtraMapFields().put(mountPath, diskProps);
        } else {
          DataNodeConfig.DiskConfig disk = new DataNodeConfig.DiskConfig(
              diskProps.get(DISK_STATE).equals(AVAILABLE_STR) ? HardwareState.AVAILABLE : HardwareState.UNAVAILABLE,
              Long.parseLong(diskProps.get(DISK_CAPACITY_STR)));
          String replicasStr = diskProps.get(REPLICAS_STR);
          if (!Utils.isNullOrEmpty(replicasStr)) {
            for (String replicaStr : replicasStr.split(REPLICAS_DELIM_STR)) {
              String[] replicaStrParts = replicaStr.split(REPLICAS_STR_SEPARATOR);
              // partition name and replica name are the same.
              String partitionName = replicaStrParts[0];
              long replicaCapacity = Long.parseLong(replicaStrParts[1]);
              String partitionClass =
                  replicaStrParts.length > 2 ? replicaStrParts[2] : clusterMapConfig.clusterMapDefaultPartitionClass;
              disk.getReplicaConfigs()
                  .put(partitionName, new DataNodeConfig.ReplicaConfig(replicaCapacity, partitionClass));
            }
          }
          dataNodeConfig.getDiskConfigs().put(mountPath, disk);
        }
      });
      return dataNodeConfig;
    }
@param instanceConfig the {@link InstanceConfig} to convert to a {@link DataNodeConfig} object.
@return the {@link DataNodeConfig}, or {@code null} if the {@link InstanceConfig} provided has an unsupported
        schema version.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/InstanceConfigToDataNodeConfigAdapter.java