code: public static SlaEventSubmitterBuilder getEventSubmitterBuilder(Dataset dataset, Optional<Job> job, FileSystem fs) {
      SlaEventSubmitterBuilder builder =
          SlaEventSubmitter.builder().datasetUrn(dataset.getUrn())
              .partition(dataset.jobProps().getProp(MRCompactor.COMPACTION_JOB_DEST_PARTITION, ""))
              .dedupeStatus(getOutputDedupeStatus(dataset.jobProps()));
  
      long previousPublishTime = getPreviousPublishTime(dataset, fs);
      long upstreamTime = dataset.jobProps().getPropAsLong(SlaEventKeys.UPSTREAM_TS_IN_MILLI_SECS_KEY, -1l);
      long recordCount = getRecordCount(job);
  
      // Previous publish only exists when this is a recompact job
      if (previousPublishTime != -1l) {
        builder.previousPublishTimestamp(Long.toString(previousPublishTime));
      }
      // Upstream time is the logical time represented by the compaction input directory
      if (upstreamTime != -1l) {
        builder.upstreamTimestamp(Long.toString(upstreamTime));
      }
      if (recordCount != -1l) {
        builder.recordCount(Long.toString(recordCount));
      }
      return builder;
    }
docs: Get an {@link SlaEventSubmitterBuilder} that has dataset urn, partition, record count, previous publish timestamp
  and dedupe status set.
  The caller MUST set eventSubmitter, eventname before submitting.