private void addUpdateResources(String dcName, Map<String, Set<String>> partitionsToInstancesInDc,
      boolean resourcesAreFullAutoCompatible) {
    HelixAdmin dcAdmin = adminForDc.get(dcName);
    List<String> instancesWithDisabledPartition = new ArrayList<>();
    HelixPropertyStore<ZNRecord> helixPropertyStore =
        helixAdminOperation == HelixAdminOperation.DisablePartition ? createHelixPropertyStore(this.clusterName,
            dataCenterToZkAddress.get(dcName).getZkConnectStrs().get(0)) : null;
    TreeMap<Integer, Set<String>> resourceIdToInstances = dcToResourceIdToInstances.get(dcName);
    Map<String, Set<Integer>> instanceNameToResources = dcToInstanceToResourceIds.get(dcName);
    Map<Integer, IdealState> resourceIdToIdealState = dcToResourceIdToIdealState.get(dcName);
    Map<String, Integer> partitionToResourceId = dcToPartitionToResourceId.get(dcName);
    // maxResource may vary from one dc to another (special partition class allows partitions to exist in one dc only)
    int maxResource = -1;
    for (int resourceId : new ArrayList<>(resourceIdToIdealState.keySet())) {
      boolean resourceModified = false;
      maxResource = Math.max(maxResource, resourceId);
      IdealState resourceIs = resourceIdToIdealState.get(resourceId);
      if (isIdealStateInFullAuto(resourceIs)) {
        info("Resource {} in dc {} is in FULL AUTO mode, ignore this resource", resourceId, dcName);
        continue;
      }
      for (String partitionName : new HashSet<>(resourceIs.getPartitionSet())) {
        Set<String> instanceSetInHelix = resourceIs.getInstanceSet(partitionName);
        Set<String> instanceSetInStatic = partitionsToInstancesInDc.remove(partitionName);
        if (instanceSetInStatic == null || instanceSetInStatic.isEmpty()) {
          info(
              "[{}] No instance set in static clustermap for partition {}, but will not remove this partition from resource. To remove partition, please use helix API",
              dcName.toUpperCase(), partitionName);
          expectMoreInHelixDuringValidate = true;
          partitionsNotForceRemovedByDc.computeIfAbsent(dcName, k -> ConcurrentHashMap.newKeySet()).add(partitionName);
          if (partitionToResourceId.get(partitionName) == resourceId) {
            info("[{}] Remove partition to resource id for partition {} and resource id{}", dcName.toUpperCase(),
                partitionName, resourceId);
            partitionToResourceId.remove(partitionName);
          }
        } else if (!instanceSetInStatic.equals(instanceSetInHelix)) {
          // we change the IdealState only when the operation is meant to bootstrap cluster or indeed update IdealState
          if (EnumSet.of(HelixAdminOperation.UpdateIdealState, HelixAdminOperation.BootstrapCluster)
              .contains(helixAdminOperation)) {
            // @formatter:off
            info(
                "[{}] Different instance sets for partition {} under resource {}. {}. "
                    + "Previous instance set: [{}], new instance set: [{}]",
                dcName.toUpperCase(), partitionName, resourceId,
                dryRun ? "No action as dry run" : "Updating Helix using static",
                String.join(",", instanceSetInHelix), String.join(",", instanceSetInStatic));
            // @formatter:on
            if (instanceSetInStatic.size() < instanceSetInHelix.size()) {
              warning("*****STOP STOP STOP*****: New instance set has less replicas than what is in helix");
            }
            if (!dryRun) {
              ArrayList<String> newInstances = new ArrayList<>(instanceSetInStatic);
              Collections.shuffle(newInstances);
              resourceIs.setPreferenceList(partitionName, newInstances);
              // Existing resources may not have ANY_LIVEINSTANCE set as the numReplicas (which allows for different
              // replication for different partitions under the same resource). So set it here (We use the name() method and
              // not the toString() method for the enum as that is what Helix uses).
              resourceIs.setReplicas(ResourceConfig.ResourceConfigConstants.ANY_LIVEINSTANCE.name());
            }
            resourceIdToInstances.computeIfAbsent(resourceId, k -> new HashSet<>()).addAll(instanceSetInStatic);
            instanceSetInStatic.forEach(
                ins -> instanceNameToResources.computeIfAbsent(ins, k -> new HashSet<>()).add(resourceId));
            resourceModified = true;
          } else if (helixAdminOperation == HelixAdminOperation.DisablePartition) {
            // if this is DisablePartition operation, we don't modify IdealState and only make InstanceConfig to disable
            // certain partition on specific node.
            // 1. extract difference between Helix and Static instance sets. Determine which replica is removed
            instanceSetInHelix.removeAll(instanceSetInStatic);
            // 2. disable removed replica on certain node.
            for (String instanceInHelixOnly : instanceSetInHelix) {
              info("Partition {} under resource {} on node {} is no longer in static clustermap. {}.", partitionName,
                  resourceId, instanceInHelixOnly, dryRun ? "No action as dry run" : "Disabling it");
              if (!dryRun) {
                InstanceConfig instanceConfig = dcAdmin.getInstanceConfig(clusterName, instanceInHelixOnly);
                String instanceName = instanceConfig.getInstanceName();
                // create a Znode (if not present) in PropertyStore for this instance before disabling partition, which
                // will be deleted in the end. The reason to create a ZNode is to ensure replica decommission is blocked
                // on updating InstanceConfig until this tool has completed. TODO, remove this logic once migration to PropertyStore is done.
                if (!instancesWithDisabledPartition.contains(instanceName)) {
                  ZNRecord znRecord = new ZNRecord(instanceName);
                  String path = PARTITION_DISABLED_ZNODE_PATH + instanceName;
                  if (!helixPropertyStore.create(path, znRecord, AccessOption.PERSISTENT)) {
                    logger.error("Failed to create a ZNode for {} in datacenter {} before disabling partition.",
                        instanceName, dcName);
                    continue;
                  }
                }
                instanceConfig.setInstanceEnabledForPartition(String.valueOf(resourceId), partitionName, false);
                dcAdmin.setInstanceConfig(clusterName, instanceInHelixOnly, instanceConfig);
                instancesWithDisabledPartition.add(instanceName);
              }
              partitionsDisabled.getAndIncrement();
            }
            // Disabling partition won't remove certain replica from IdealState. So replicas of this partition in Helix
            // will be more than that in static clustermap.
            expectMoreInHelixDuringValidate = true;
          }
        } else {
          resourceIdToInstances.computeIfAbsent(resourceId, k -> new HashSet<>()).addAll(instanceSetInStatic);
          instanceSetInStatic.forEach(
              ins -> instanceNameToResources.computeIfAbsent(ins, k -> new HashSet<>()).add(resourceId));
        }
      }
      // update state model def if necessary
      if (!resourceIs.getStateModelDefRef().equals(stateModelDef)) {
        info("[{}] Resource {} has different state model {}. Updating it with {}", dcName.toUpperCase(), resourceId,
            resourceIs.getStateModelDefRef(), stateModelDef);
        resourceIs.setStateModelDefRef(stateModelDef);
        resourceModified = true;
      }
      resourceIs.setNumPartitions(resourceIs.getPartitionSet().size());
      if (resourceModified) {
        if (resourceIs.getPartitionSet().isEmpty()) {
          info("[{}] Resource {} has no partition, to remove this resource, please use helix API", dcName.toUpperCase(),
              resourceId);
          resourceIdToIdealState.remove(resourceId);
          resourceIdToInstances.remove(resourceId);
          expectMoreInHelixDuringValidate = true;
        } else {
          if (!dryRun) {
            dcAdmin.setResourceIdealState(clusterName, String.valueOf(resourceId), resourceIs);
            System.out.println("------------------add resource!");
            System.out.println(resourceId);
          }
          resourcesUpdated.getAndIncrement();
        }
      }
    }
    // note that disabling partition also updates InstanceConfig of certain nodes which host the partitions.
    instancesUpdated.getAndAdd(instancesWithDisabledPartition.size());
    // if there are some partitions are disabled, mark whole disabling process complete in Helix PropertyStore to unblock
    // replica decommission thread on each datanode.
    if (helixPropertyStore != null) {
      maybeAwaitForLatch();
      for (String instanceName : instancesWithDisabledPartition) {
        String path = PARTITION_DISABLED_ZNODE_PATH + instanceName;
        if (!helixPropertyStore.remove(path, AccessOption.PERSISTENT)) {
          logger.error("Failed to remove a ZNode for {} in datacenter {} after disabling partition completed.",
              instanceName, dcName);
        }
      }
      helixPropertyStore.stop();
    }

    // When the cluster is empty and we have maxInstancesInOneResourceForFullAuto greater than 0, we would by default create
    // resources in FULL_AUTO compatible mode.
    Set<String> allPartitionsInIdealState = dcToPartitionToResourceId.get(dcName).keySet();
    partitionsToInstancesInDc.keySet().removeAll(allPartitionsInIdealState);
    if (resourcesAreFullAutoCompatible) {
      info(
          "[{}] Resources are full auto compatible, will create new partitions in resources that are also full auto compatible",
          dcName.toUpperCase());
      Map<String, Integer> instanceNameToOneResource = new HashMap<>();
      for (Map.Entry<String, Set<Integer>> ent : instanceNameToResources.entrySet()) {
        // instance can belong to multiple resources, but each resource should have different number of replicas, so
        // (instance, numReplica) should point to one resource
        for (int resourceId : ent.getValue()) {
          int numReplica = numReplicasForIdealState(dcName, resourceIdToIdealState.get(resourceId));
          instanceNameToOneResource.put(ent.getKey() + "_" + numReplica, resourceId);
        }
      }
      // Partitions that not are in the helix yet, there are two different cases.
      // The instances this partition is assigned to already in the helix resources, then this partition should go to the same resource.
      // The instances this partition is assigned to don't yet exist in helix resources, then this partition could go to a new resource.
      updateIdealStateWithNewPartitionsForFullAuto(dcName, partitionsToInstancesInDc, instanceNameToOneResource,
          resourceIdToInstances, resourceIdToIdealState);
      if (!partitionsToInstancesInDc.isEmpty()) {
        createOrUpdateIdealStateForFullAuto(dcName, partitionsToInstancesInDc, resourceIdToInstances);
      }
    } else {
      // Add what is not already in Helix under new resources.
      int fromIndex = 0;
      List<Map.Entry<String, Set<String>>> newPartitions = new ArrayList<>(partitionsToInstancesInDc.entrySet());
      while (fromIndex < newPartitions.size()) {
        String resourceName = Integer.toString(++maxResource);
        int toIndex = Math.min(fromIndex + maxPartitionsInOneResource, newPartitions.size());
        List<Map.Entry<String, Set<String>>> partitionsUnderNextResource = newPartitions.subList(fromIndex, toIndex);
        fromIndex = toIndex;
        buildAndCreateIdealState(dcName, resourceName, partitionsUnderNextResource);
      }
    }
  }
Add and/or update resources in Helix based on the information in the static cluster map. This may involve adding
or removing partitions from under a resource, and adding or dropping resources altogether. This may also involve
changing the instance set for a partition under a resource, based on the static cluster map.
@param dcName the name of the datacenter being processed.
@param partitionsToInstancesInDc a map to be filled with the mapping of partitions to their instance sets in the
                                 given datacenter.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/HelixBootstrapUpgradeUtil.java