private Pair<List<ReplicaId>, List<ReplicaId>> updateInstanceInfo(DataNodeConfig dataNodeConfig, String dcName)
        throws Exception {
      final List<ReplicaId> addedReplicas = new ArrayList<>();
      final List<ReplicaId> removedReplicas = new ArrayList<>();
      String instanceName = dataNodeConfig.getInstanceName();
      logger.info("Updating replicas info for existing node {}", instanceName);
      Set<String> sealedReplicas = dataNodeConfig.getSealedReplicas();
      Set<String> partiallySealedReplicas = dataNodeConfig.getPartiallySealedReplicas();
      Set<String> stoppedReplicas = dataNodeConfig.getStoppedReplicas();
      AmbryDataNode dataNode = instanceNameToAmbryDataNode.get(instanceName);
      ConcurrentHashMap<String, AmbryReplica> currentReplicasOnNode = ambryDataNodeToAmbryReplicas.get(dataNode);
      ConcurrentHashMap<String, AmbryReplica> replicasFromInstanceConfig = new ConcurrentHashMap<>();
      Map<String, AmbryDisk> mountPathToDisk = ambryDataNodeToAmbryDisks.get(dataNode)
          .stream()
          .collect(Collectors.toMap(AmbryDisk::getMountPath, disk -> disk));
      Map<AmbryPartition, List<AmbryReplica>> replicaToAddByPartition = new HashMap<>();
      for (Map.Entry<String, DataNodeConfig.DiskConfig> diskEntry : dataNodeConfig.getDiskConfigs().entrySet()) {
        String mountPath = diskEntry.getKey();
        DataNodeConfig.DiskConfig diskConfig = diskEntry.getValue();
        AmbryDisk disk = mountPathToDisk.getOrDefault(mountPath, null);
        if (disk == null) {
          logger.warn("{} is a new disk or unrecognizable disk which is not supported on existing node {}.", mountPath,
              instanceName);
          // TODO support dynamically adding disk in the future
          continue;
        }
        // update disk capacity if needed
        if (disk.getRawCapacityInBytes() != diskConfig.getDiskCapacityInBytes()) {
          long prevDiskCapacity = disk.getRawCapacityInBytes();
          logger.info("Capacity of disk at {} on {} has changed. Previous was: {} bytes, new capacity is {} bytes",
              mountPath, instanceName, prevDiskCapacity, diskConfig.getDiskCapacityInBytes());
          disk.setDiskCapacityInBytes(diskConfig.getDiskCapacityInBytes());
          addClusterWideRawCapacity(diskConfig.getDiskCapacityInBytes() - prevDiskCapacity);
        }
        if (disk.getState() != diskConfig.getState()) {
          logger.info("State of disk at {} on {} has changed, Previous was: {}, new stats is {}", mountPath,
              instanceName, disk.getState(), diskConfig.getState());
          disk.setState(diskConfig.getState());
        }
        for (Map.Entry<String, DataNodeConfig.ReplicaConfig> replicaEntry : diskConfig.getReplicaConfigs().entrySet()) {
          // partition name and replica name are the same.
          String partitionName = replicaEntry.getKey();
          DataNodeConfig.ReplicaConfig replicaConfig = replicaEntry.getValue();
          if (currentReplicasOnNode.containsKey(partitionName)) {
            // if replica is already present
            AmbryReplica existingReplica = currentReplicasOnNode.get(partitionName);
            // 1. directly add it into "replicasFromInstanceConfig" map
            replicasFromInstanceConfig.put(partitionName, existingReplica);
            // 2. update replica seal/stop state
            updateReplicaStateAndOverrideIfNeeded(existingReplica, sealedReplicas, partiallySealedReplicas,
                stoppedReplicas);
          } else {
            // if this is a new replica and doesn't exist on node
            logger.info("Adding new replica {} to existing node {} in {}", partitionName, instanceName, dcName);
            // this can be a brand new partition that is added to an existing node
            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 new replica belonging to this partition or find the existing replica from bootstrapReplicas map.
            AmbryReplica replica;
            if (selfInstanceName.equals(instanceName)) {
              // if this is a newly added replica on current instance, it should be present in bootstrapReplicas map.
              replica = fetchBootstrapReplica(mappedPartition.toPathString());
              if (replica == null) {
                logger.error("Replica {} is not present in bootstrap replica set, abort instance info update",
                    mappedPartition.toPathString());
                throw new IllegalStateException("Replica to add is not present in bootstrap replica map");
              }
            } else {
              replica = new AmbryServerReplica(clusterMapConfig, mappedPartition, disk,
                  stoppedReplicas.contains(partitionName), replicaConfig.getReplicaCapacityInBytes(),
                  resolveReplicaSealStatus(partitionName, sealedReplicas, partiallySealedReplicas));
            }
            updateReplicaStateAndOverrideIfNeeded(replica, sealedReplicas, partiallySealedReplicas, stoppedReplicas);
            // add new created replica to "replicasFromInstanceConfig" map
            replicasFromInstanceConfig.put(partitionName, replica);
            // Put new replica into partition-to-replica map temporarily (this is to avoid any exception thrown within the
            // loop before updating "ambryDataNodeToAmbryReplicas" map. If we update call addReplicasToPartition here
            // immediately, the exception may cause inconsistency between "ambryPartitionToAmbryReplicas" and
            // "ambryDataNodeToAmbryReplicas")
            replicaToAddByPartition.put(mappedPartition, Collections.singletonList(replica));
          }
        }
      }
      // update "ambryDataNodeToAmbryReplicas" map and "addedReplicas" list
      replicaToAddByPartition.forEach((k, v) -> {
        addReplicasToPartition(k, v);
        addedReplicas.addAll(v);
      });
      // update ambryDataNodeToAmbryReplicas map by adding "replicasFromInstanceConfig"
      ambryDataNodeToAmbryReplicas.put(instanceNameToAmbryDataNode.get(instanceName), replicasFromInstanceConfig);
      // Derive old replicas that are removed and delete them from partition
      currentReplicasOnNode.keySet()
          .stream()
          .filter(partitionName -> !replicasFromInstanceConfig.containsKey(partitionName))
          .forEach(pName -> {
            logger.info("Removing replica {} from existing node {}", pName, instanceName);
            AmbryReplica ambryReplica = currentReplicasOnNode.get(pName);
            removeReplicasFromPartition(ambryReplica.getPartitionId(), Collections.singletonList(ambryReplica));
            removedReplicas.add(ambryReplica);
          });
      return new Pair<>(addedReplicas, removedReplicas);
    }
Update info of an existing instance. This may happen in following cases: (1) new replica is added; (2) old replica
is removed; (3) replica's state has changed (i.e. becomes seal/unseal).
@param dataNodeConfig the {@link DataNodeConfig} used to update info of instance.
@param dcName
@return a pair of lists: (1) new added replicas; (2) removed old replicas, during this update.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/HelixClusterManager.java