protected synchronized void purgeJob ( KillJobAction action ) throws IOException { JobID jobId = action . getJobID ( ) ; LOG . info ( "Received 'KillJobAction' for job: " + jobId ) ; RunningJob rjob = null ; synchronized ( runningJobs ) { rjob = runningJobs . get ( jobId ) ; } if ( rjob == null ) { if ( LOG . isDebugEnabled ( ) ) { // We cleanup the job on all tasktrackers in the cluster // so there is a good chance it never ran a single task from it LOG . debug ( "Unknown job " + jobId + " being deleted." ) ; } } else { synchronized ( rjob ) { // Add this tips of this job to queue of tasks to be purged for ( TaskInProgress tip : rjob . tasks ) { tip . jobHasFinished ( false ) ; Task t = tip . getTask ( ) ; if ( t . isMapTask ( ) ) { indexCache . removeMap ( tip . getTask ( ) . getTaskID ( ) . toString ( ) ) ; } // Remove it from the runningTasks if ( this . runningTasks . containsKey ( t . getTaskID ( ) ) ) { LOG . info ( "Remove " + t . getTaskID ( ) + " from runningTask by purgeJob" ) ; this . runningTasks . remove ( t . getTaskID ( ) ) ; } } // Delete the job directory for this // task if the job is done/failed if ( ! rjob . keepJobFiles ) { PathDeletionContext [ ] contexts = buildPathDeletionContexts ( localFs , getLocalFiles ( fConf , getLocalJobDir ( rjob . getJobID ( ) . toString ( ) ) ) ) ; directoryCleanupThread . addToQueue ( contexts ) ; } // Remove this job rjob . tasks . clear ( ) ; } } synchronized ( runningJobs ) { runningJobs . remove ( jobId ) ; } }
The task tracker is done with this job so we need to clean up .