Pair<AggregatedPartitionClassStorageStats, AggregatedPartitionClassStorageStats> aggregateHostPartitionClassStorageStatsWrappers(
      PartitionClassStorageStatsIterator partitionClassStorageStatsIterator) throws IOException {
    Map<String, Map<Long, Map<Short, Map<Short, ContainerStorageStats>>>> combinedHostPartitionClassStorageStatsMap =
        new HashMap<>();
    Map<String, Map<Long, Map<Short, Map<Short, ContainerStorageStats>>>> selectedHostPartitionClassStorageStatsMap =
        new HashMap<>();
    Map<Long, Long> partitionTimestampMap = new HashMap<>();
    Map<Long, Long> partitionPhysicalStorageMap = new HashMap<>();

    while (partitionClassStorageStatsIterator.hasNext()) {
      Pair<String, HostPartitionClassStorageStatsWrapper> statsWrapperEntry = partitionClassStorageStatsIterator.next();
      if (statsWrapperEntry.getSecond() == null) {
        continue;
      }
      String hostname = statsWrapperEntry.getFirst();
      HostPartitionClassStorageStatsWrapper hostPartitionClassStorageStatsWrapper = statsWrapperEntry.getSecond();
      HostPartitionClassStorageStats hostPartitionClassStorageStats = hostPartitionClassStorageStatsWrapper.getStats();
      HostPartitionClassStorageStats hostPartitionClassStorageStatsCopy1 =
          new HostPartitionClassStorageStats(hostPartitionClassStorageStats);
      HostPartitionClassStorageStats hostPartitionClassStorageStatsCopy2 =
          new HostPartitionClassStorageStats(hostPartitionClassStorageStats);
      combineRawHostPartitionClassStorageStatsMap(combinedHostPartitionClassStorageStatsMap,
          hostPartitionClassStorageStatsCopy1.getStorageStats());
      selectRawHostPartitionClassStorageStatsMap(selectedHostPartitionClassStorageStatsMap,
          hostPartitionClassStorageStatsCopy2.getStorageStats(), partitionTimestampMap, partitionPhysicalStorageMap,
          hostPartitionClassStorageStatsWrapper.getHeader().getTimestamp(), hostname);
    }

    if (logger.isTraceEnabled()) {
      logger.trace("Combined raw HostPartitionClassStorageStats {}",
          mapper.writeValueAsString(combinedHostPartitionClassStorageStatsMap));
      logger.trace("Selected raw HostPartitionClassStorageStats {}",
          mapper.writeValueAsString(selectedHostPartitionClassStorageStatsMap));
    }

    AggregatedPartitionClassStorageStats combinedAggregated = new AggregatedPartitionClassStorageStats(
        aggregateHostPartitionClassStorageStats(combinedHostPartitionClassStorageStatsMap));
    AggregatedPartitionClassStorageStats selectedAggregated = new AggregatedPartitionClassStorageStats(
        aggregateHostPartitionClassStorageStats(selectedHostPartitionClassStorageStatsMap));
    if (logger.isTraceEnabled()) {
      logger.trace("Aggregated combined {}", mapper.writeValueAsString(combinedAggregated));
      logger.trace("Aggregated selected {}", mapper.writeValueAsString(selectedAggregated));
    }
    return new Pair<>(combinedAggregated, selectedAggregated);
  }
Aggregate all {@link HostPartitionClassStorageStatsWrapper} to generate two {@link AggregatedPartitionClassStorageStats}s.
First {@link AggregatedPartitionClassStorageStats} is the sum of all {@link HostPartitionClassStorageStatsWrapper}s.
Second {@link AggregatedPartitionClassStorageStats} is an aggregated storage stats for all replicas of each partition.
@param partitionClassStorageStatsIterator An iterator for PartitionClassStorageStatsIterator
@return A {@link Pair} of {@link AggregatedPartitionClassStorageStats}.
@throws IOException
ambry-clustermap/src/main/java/com/github/ambry/clustermap/MySqlClusterAggregator.java