code: private void cleanupStagingData(JobState jobState)
        throws JobException {
      if (jobState.getPropAsBoolean(ConfigurationKeys.CLEANUP_STAGING_DATA_BY_INITIALIZER, false)) {
        //Clean up will be done by initializer.
        return;
      }
  
      try {
        if (!canCleanStagingData(jobState)) {
          LOG.error("Job " + jobState.getJobName() + " has unfinished commit sequences. Will not clean up staging data.");
          return;
        }
      } catch (IOException e) {
        throw new JobException("Failed to check unfinished commit sequences", e);
      }
  
      if (this.jobContext.shouldCleanupStagingDataPerTask()) {
        cleanupStagingDataPerTask(jobState);
      } else {
        cleanupStagingDataForEntireJob(jobState);
      }
    }
docs: Cleanup the job's task staging data. This is not doing anything in case job succeeds
  and data is successfully committed because the staging data has already been moved
  to the job output directory. But in case the job fails and data is not committed,
  we want the staging data to be cleaned up.
  
  Property {@link ConfigurationKeys#CLEANUP_STAGING_DATA_PER_TASK} controls whether to cleanup
  staging data per task, or to cleanup entire job's staging data at once.
  
  Staging data will not be cleaned if the job has unfinished {@link CommitSequence}s.