code: private void submitSlaEvent(Job job) {
      try {
        CompactionSlaEventHelper
            .getEventSubmitterBuilder(this.dataset, Optional.of(job), this.fs)
            .eventSubmitter(this.eventSubmitter)
            .eventName(CompactionSlaEventHelper.COMPACTION_COMPLETED_EVENT_NAME)
            .additionalMetadata(
                CompactionSlaEventHelper.LATE_RECORD_COUNT,
                Long.toString(this.lateOutputRecordCountProvider.getRecordCount(this.getApplicableFilePaths(this.dataset
                    .outputLatePath(), this.fs))))
            .additionalMetadata(
                CompactionSlaEventHelper.REGULAR_RECORD_COUNT,
                Long.toString(this.outputRecordCountProvider.getRecordCount(this.getApplicableFilePaths(this.dataset
                    .outputPath(), this.fs))))
            .additionalMetadata(CompactionSlaEventHelper.RECOMPATED_METADATA_NAME,
                Boolean.toString(this.dataset.needToRecompact())).build().submit();
      } catch (Throwable e) {
        LOG.warn("Failed to submit compaction completed event:" + e, e);
      }
    }
docs: Submit an event when compaction MR job completes