private List<ReplicaId> initializeDisksAndReplicasOnNode(AmbryDataNode datanode, DataNodeConfig dataNodeConfig)
        throws Exception {
      List<ReplicaId> addedReplicas = new ArrayList<>();
      Set<String> sealedReplicas = dataNodeConfig.getSealedReplicas();
      Set<String> partiallySealedReplicas = dataNodeConfig.getPartiallySealedReplicas();
      Set<String> stoppedReplicas = dataNodeConfig.getStoppedReplicas();
      ambryDataNodeToAmbryReplicas.put(datanode, new ConcurrentHashMap<>());
      ambryDataNodeToAmbryDisks.put(datanode, ConcurrentHashMap.newKeySet());
      for (Map.Entry<String, DataNodeConfig.DiskConfig> diskEntry : dataNodeConfig.getDiskConfigs().entrySet()) {
        String mountPath = diskEntry.getKey();
        DataNodeConfig.DiskConfig diskConfig = diskEntry.getValue();

        // Create disk
        AmbryDisk disk = new AmbryDisk(clusterMapConfig, datanode, mountPath, diskConfig.getState(),
            diskConfig.getDiskCapacityInBytes());
        ambryDataNodeToAmbryDisks.get(datanode).add(disk);
        addClusterWideRawCapacity(diskConfig.getDiskCapacityInBytes());

        for (Map.Entry<String, DataNodeConfig.ReplicaConfig> replicaEntry : diskConfig.getReplicaConfigs().entrySet()) {
          String partitionName = replicaEntry.getKey();
          DataNodeConfig.ReplicaConfig replicaConfig = replicaEntry.getValue();

          AmbryPartition mappedPartition =
              new AmbryPartition(Long.parseLong(partitionName), replicaConfig.getPartitionClass(),
                  helixClusterManagerQueryHelper);
          // Ensure only one AmbryPartition instance exists for specific partition.
          mappedPartition = addPartitionIfAbsent(mappedPartition, replicaConfig.getReplicaCapacityInBytes());
          ensurePartitionAbsenceOnNodeAndValidateCapacity(mappedPartition, datanode,
              replicaConfig.getReplicaCapacityInBytes());
          // Create replica associated with this node and this partition
          AmbryReplica replica =
              new AmbryServerReplica(clusterMapConfig, mappedPartition, disk, stoppedReplicas.contains(partitionName),
                  replicaConfig.getReplicaCapacityInBytes(),
                  resolveReplicaSealStatus(partitionName, sealedReplicas, partiallySealedReplicas));
          ambryDataNodeToAmbryReplicas.get(datanode).put(mappedPartition.toPathString(), replica);
          addReplicasToPartition(mappedPartition, Collections.singletonList(replica));
          addedReplicas.add(replica);
        }
      }
      return addedReplicas;
    }
Initialize the disks and replicas on the given node. Create partitions if this is the first time a replica of
that partition is being constructed. If partition override is enabled, the seal state of replica is determined by
partition info in HelixPropertyStore, if disabled, the seal state is determined by {@code dataNodeConfig}.
@param datanode the {@link AmbryDataNode} that is being initialized.
@param dataNodeConfig the {@link DataNodeConfig} associated with this datanode.
@return a list of newly added replicas on this node.
@throws Exception if creation of {@link AmbryDisk} throws an Exception.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/HelixClusterManager.java