code: private void aggreagteJobStats(JobDetails jobDetails, byte[] rowKey,
        Context context, AggregationConstants.AGGREGATION_TYPE aggType)
        throws IOException, InterruptedException {
  
      byte[] aggStatusCol = null;
      switch (aggType) {
      case DAILY:
        aggStatusCol =
            AggregationConstants.JOB_DAILY_AGGREGATION_STATUS_COL_BYTES;
        break;
      case WEEKLY:
        aggStatusCol =
            AggregationConstants.JOB_WEEKLY_AGGREGATION_STATUS_COL_BYTES;
        break;
      default:
        LOG.error("Unknown aggregation type " + aggType);
        return;
      }
  
      boolean aggStatus =
          appSummaryService.aggregateJobDetails(jobDetails, aggType);
      context.progress();
      LOG.debug("Status of aggreagting stats for " + aggType + "=" + aggStatus);
      if (aggStatus) {
        // update raw table for this history file with aggregation status
        // Indicate that we processed the agg for this RAW successfully
        // so that we can skip it on the next scan (or not).
        Put aggStatusPut =
            rawService.getAggregatedStatusPut(rowKey, aggStatusCol, aggStatus);
        // TODO
        // In the unlikely event of multiple mappers running against one RAW
        // row, with one succeeding and one failing,
        // there could be a race where the
        // raw does not properly indicate the true status
        // (which is questionable in
        // any case with multiple simultaneous runs with different outcome).
        context.write(RAW_TABLE, aggStatusPut);
      }
    }
docs: aggregate this job's stats only if re-aggregation is turned on OR
  aggreation is on AND job not already aggregated
  
  if job has already been aggregated, we don't want to mistakenly aggregate
  again