code: private void submitRecordsCountsEvent() {
      long lateOutputRecordCount = this.datasetHelper.getLateOutputRecordCount();
      long outputRecordCount = this.datasetHelper.getOutputRecordCount();
  
      try {
        CompactionSlaEventHelper
            .getEventSubmitterBuilder(this.dataset, Optional.<Job> absent(), this.fs)
            .eventSubmitter(this.eventSubmitter)
            .eventName(CompactionSlaEventHelper.COMPACTION_RECORD_COUNT_EVENT)
            .additionalMetadata(CompactionSlaEventHelper.DATASET_OUTPUT_PATH, this.dataset.outputPath().toString())
            .additionalMetadata(
                CompactionSlaEventHelper.LATE_RECORD_COUNT,
                Long.toString(lateOutputRecordCount))
            .additionalMetadata(
                CompactionSlaEventHelper.REGULAR_RECORD_COUNT,
                Long.toString(outputRecordCount))
            .additionalMetadata(CompactionSlaEventHelper.NEED_RECOMPACT, Boolean.toString(this.dataset.needToRecompact()))
            .build().submit();
      } catch (Throwable e) {
        LOG.warn("Failed to submit late event count:" + e, e);
      }
    }
docs: Submit an event reporting late record counts and non-late record counts.