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);
    }
  }
private void cleanupStagingData ( JobState jobState ) throws JobException { if ( jobState . getPropAsBoolean ( ConfigurationKeys . CLEANUP_STAGING_DATA_BY_INITIALIZER , false ) ) { 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 ) ; } }
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.