private void cleanLeftoverStagingData ( WorkUnitStream workUnits , 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 ) ; } try { if ( this . jobContext . shouldCleanupStagingDataPerTask ( ) ) { if ( workUnits . isSafeToMaterialize ( ) ) { Closer closer = Closer . create ( ) ; Map < String , ParallelRunner > parallelRunners = Maps . newHashMap ( ) ; try { for ( WorkUnit workUnit : JobLauncherUtils . flattenWorkUnits ( workUnits . getMaterializedWorkUnitCollection ( ) ) ) { JobLauncherUtils . cleanTaskStagingData ( new WorkUnitState ( workUnit , jobState ) , LOG , closer , parallelRunners ) ; } } catch ( Throwable t ) { throw closer . rethrow ( t ) ; } finally { closer . close ( ) ; } } else { throw new RuntimeException ( "Work unit streams do not support cleaning staging data per task." ) ; } } else { if ( jobState . getPropAsBoolean ( ConfigurationKeys . CLEANUP_OLD_JOBS_DATA , ConfigurationKeys . DEFAULT_CLEANUP_OLD_JOBS_DATA ) ) { JobLauncherUtils . cleanUpOldJobData ( jobState , LOG , jobContext . getStagingDirProvided ( ) , jobContext . getOutputDirProvided ( ) ) ; } JobLauncherUtils . cleanJobStagingData ( jobState , LOG ) ; } } catch ( Throwable t ) { // Catch Throwable instead of just IOException to make sure failure of this won't affect the current run LOG . error ( "Failed to clean leftover staging data" , t ) ; } }
Cleanup the left - over staging data possibly from the previous run of the job that may have failed and not cleaned up its staging data .