@Override
    public void run() {
      logger.info("Aggregating partition class stats for local report");
      try {
        long totalFetchAndAggregateStartTimeMs = time.milliseconds();
        List<PartitionId> unreachablePartitions = new ArrayList<>();
        Map<String, Map<Long, Map<Short, Map<Short, ContainerStorageStats>>>> hostPartitionClassStorageStatsMap =
            new HashMap<>();

        // 1. First, collect partition class stats from each replicas and aggregate the result to aggregatedSnapshot.
        Set<PartitionId> partitionIds = new HashSet<>(partitionToReplicaMap.keySet());
        Iterator<PartitionId> iterator = partitionIds.iterator();
        while (!cancelled && iterator.hasNext()) {
          PartitionId partitionId = iterator.next();
          logger.debug("Aggregating partition class stats for local report started for store {}", partitionId);
          collectAndAggregatePartitionClassStorageStats(hostPartitionClassStorageStatsMap, partitionId,
              unreachablePartitions);
        }

        // 2. Second, filter out the unreachable partitions that are not in the map.
        List<String> unreachableStores = examineUnreachablePartitions(unreachablePartitions);
        if (!cancelled) {
          metrics.totalFetchAndAggregateTimeMs.update(time.milliseconds() - totalFetchAndAggregateStartTimeMs);
          // 3. Construct a StatsWrapper.
          StatsHeader statsHeader =
              new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, time.milliseconds(), partitionIds.size(),
                  partitionIds.size() - unreachableStores.size(), unreachableStores);
          HostPartitionClassStorageStatsWrapper statsWrapper = new HostPartitionClassStorageStatsWrapper(statsHeader,
              new HostPartitionClassStorageStats(hostPartitionClassStorageStatsMap));
          // 4. Persist this statsWrapper to mysql database.
          accountStatsStore.storeHostPartitionClassStorageStats(statsWrapper);
        }
      } catch (Exception e) {
        metrics.statsAggregationFailureCount.inc();
        logger.error("Exception while aggregating partition class stats for mysql", e);
      }
    }
Runnable class that collects, publishes partition class stats to mysql database.
ambry-server/src/main/java/com/github/ambry/server/StatsManager.java