@Override
  public TaskResult run() {
    Exception exception = null;
    Histogram fetchTimeMs = statsReportType == StatsReportType.ACCOUNT_REPORT ? metrics.accountStatsFetchTimeMs
        : metrics.partitionClassStatsFetchTimeMs;
    Histogram aggregationTimeMs =
        statsReportType == StatsReportType.ACCOUNT_REPORT ? metrics.accountStatsAggregationTimeMs
            : metrics.partitionClassStatsAggregationTimeMs;
    long startTimeMs = System.currentTimeMillis();
    AggregatedAccountStorageStats aggregatedAccountStorageStats = null;
    try {
      List<String> instanceNames = manager.getClusterManagmentTool().getInstancesInCluster(manager.getClusterName());
      if (statsReportType == StatsReportType.ACCOUNT_REPORT) {
        logger.info("Aggregating stats from " + instanceNames.size() + " hosts");
        Pair<AggregatedAccountStorageStats, AggregatedAccountStorageStats> results =
            clusterAggregator.aggregateHostAccountStorageStatsWrappers(new AccountStorageStatsIterator(instanceNames, accountStatsStore, clusterMapConfig));
        if (clusterMapConfig.clustermapEnableDeleteInvalidDataInMysqlAggregationTask) {
          removeInvalidAggregatedAccountAndContainerStats(results.getSecond());
        }
        accountStatsStore.storeAggregatedAccountStorageStats(results.getSecond());
        aggregatedAccountStorageStats = results.getFirst();
      } else if (statsReportType == StatsReportType.PARTITION_CLASS_REPORT) {
        logger.info("Aggregating stats from " + instanceNames.size() + " hosts");
        Pair<AggregatedPartitionClassStorageStats, AggregatedPartitionClassStorageStats> results =
            clusterAggregator.aggregateHostPartitionClassStorageStatsWrappers(new PartitionClassStorageStatsIterator(instanceNames, accountStatsStore, clusterMapConfig));
        if (clusterMapConfig.clustermapEnableDeleteInvalidDataInMysqlAggregationTask) {
          removeInvalidAggregatedPartitionClassStats(results.getSecond());
        }
        accountStatsStore.storeAggregatedPartitionClassStorageStats(results.getSecond());
      }

      // Create a base report at the beginning of each month.
      // Check if there is a base report for this month or not.
      if (clusterMapConfig.clustermapEnableAggregatedMonthlyAccountReport
          && statsReportType == StatsReportType.ACCOUNT_REPORT) {
        // Get the month, if not the same month, then copy the aggregated stats and update the month
        String currentMonthValue =
            LocalDateTime.ofEpochSecond(time.seconds(), 0, ZONE_OFFSET).format(TIMESTAMP_FORMATTER);
        String recordedMonthValue = accountStatsStore.queryRecordedMonth();
        if (recordedMonthValue == null || recordedMonthValue.isEmpty() || !currentMonthValue.equals(
            recordedMonthValue)) {
          if (clusterMapConfig.clustermapEnableDeleteInvalidDataInMysqlAggregationTask) {
            accountStatsStore.deleteSnapshotOfAggregatedAccountStats();
          }
          logger.info("Taking snapshot of aggregated stats for month " + currentMonthValue);
          accountStatsStore.takeSnapshotOfAggregatedAccountStatsAndUpdateMonth(currentMonthValue);
        }
      }
      aggregationTimeMs.update(System.currentTimeMillis() - startTimeMs);
      return new TaskResult(TaskResult.Status.COMPLETED, "Aggregation success");
    } catch (Exception e) {
      logger.error("Exception thrown while aggregating stats from container stats reports across all nodes ", e);
      exception = e;
      return new TaskResult(TaskResult.Status.FAILED, "Exception thrown");
    } finally {
      if (clusterMapConfig.clustermapEnableContainerDeletionAggregation && callback != null
          && aggregatedAccountStorageStats != null && statsReportType.equals(StatsReportType.ACCOUNT_REPORT)) {
        callback.onCompletion(aggregatedAccountStorageStats, exception);
      }
    }
  }
Instantiates {@link MySqlReportAggregatorTask}.
@param manager The {@link HelixManager} to retrieve all storage nodes.
@param relevantTimePeriodInMs relevant time period in ms within which values are considered to be valid. Values
                              outside of this period will be ignored.
@param statsReportType the type of stats report
@param accountStatsStore The {@link AccountStatsStore} to retrieve stats and store aggregated stats.
@param callback a callback which will be invoked when the aggregation report has been generated successfully.
@param clusterMapConfig the {@link ClusterMapConfig} associated with helix participant.
@param registry the {@link MetricRegistry}.
ambry-clustermap/src/main/java/com/github/ambry/clustermap/MySqlReportAggregatorTask.java