@Override
  public void storeHostAccountStorageStats(HostAccountStorageStatsWrapper statsWrapper) throws Exception {
    AccountReportsDao.StorageBatchUpdater batch = accountReportsDao.new StorageBatchUpdater(config.updateBatchSize);
    int batchSize = 0;
    long startTimeMs = System.currentTimeMillis();
    HostAccountStorageStats prevHostStats = Optional.ofNullable(previousHostAccountStorageStatsWrapper)
        .map(HostAccountStorageStatsWrapper::getStats)
        .orElseGet(HostAccountStorageStats::new);
    HostAccountStorageStats currHostStats = statsWrapper.getStats();
    // Find the differences between two {@link HostAccountStorageStats} and apply them to the given {@link ContainerStorageStatsFunction}.
    // The difference is defined as
    // 1. If a container storage usage exists in both HostAccountStorageStats, and the values are different.
    // 2. If a container storage usage only exists in current HostAccountStorageStats.
    // If a container storage usage only exists in the previous HostAccountStorageStats, then it will not be applied to the given function.
    Map<Long, Map<Short, Map<Short, ContainerStorageStats>>> currPartitionMap = currHostStats.getStorageStats();
    Map<Long, Map<Short, Map<Short, ContainerStorageStats>>> prevPartitionMap = prevHostStats.getStorageStats();
    for (Map.Entry<Long, Map<Short, Map<Short, ContainerStorageStats>>> currPartitionMapEntry : currPartitionMap.entrySet()) {
      long partitionId = currPartitionMapEntry.getKey();
      Map<Short, Map<Short, ContainerStorageStats>> currAccountMap = currPartitionMapEntry.getValue();
      Map<Short, Map<Short, ContainerStorageStats>> prevAccountMap =
          prevPartitionMap.getOrDefault(partitionId, Collections.emptyMap());
      // It's possible that this accountStatsSnapshot has empty submap, if all stats snapshots in the submap have 0
      // as its value.
      for (Map.Entry<Short, Map<Short, ContainerStorageStats>> currAccountMapEntry : currAccountMap.entrySet()) {
        short accountId = currAccountMapEntry.getKey();
        Map<Short, ContainerStorageStats> currContainerMap = currAccountMapEntry.getValue();
        Map<Short, ContainerStorageStats> prevContainerMap =
            prevAccountMap.getOrDefault(accountId, Collections.emptyMap());
        for (Map.Entry<Short, ContainerStorageStats> currContainerMapEntry : currContainerMap.entrySet()) {
          short containerId = currContainerMapEntry.getKey();
          ContainerStorageStats currContainerStorageStats = currContainerMapEntry.getValue();
          if (!prevContainerMap.containsKey(containerId) || !currContainerStorageStats.equals(
              prevContainerMap.get(containerId))) {
            batch.addUpdateToBatch(clusterName, hostname, (int) partitionId, accountId, currContainerStorageStats);
            batchSize++;
          }
        }
      }
    }
    batch.flush();
    storeMetrics.batchSize.update(batchSize);
    storeMetrics.insertAccountStatsTimeMs.update(System.currentTimeMillis() - startTimeMs);

    deleteContainerAccountStats(prevPartitionMap, currPartitionMap);
    storeMetrics.publishTimeMs.update(System.currentTimeMillis() - startTimeMs);
    previousHostAccountStorageStatsWrapper = statsWrapper;
    writeStatsToLocalBackupFile();
  }
Store the {@link HostAccountStorageStatsWrapper} to mysql database. This method ignores the error information from {@link HostAccountStorageStatsWrapper}
and only publish the container storage usages that are different from the previous one.
@param statsWrapper The {@link HostAccountStorageStatsWrapper} to publish.
ambry-mysql/src/main/java/com/github/ambry/accountstats/AccountStatsMySqlStore.java