public void incompleteSubTask ( TaskAttemptID taskid , JobStatus jobStatus ) { // // Note the failure and its location // TaskStatus status = taskStatuses . get ( taskid ) ; String trackerName ; String trackerHostName = null ; TaskStatus . State taskState = TaskStatus . State . FAILED ; if ( status != null ) { trackerName = status . getTaskTracker ( ) ; trackerHostName = JobInProgressTraits . convertTrackerNameToHostName ( trackerName ) ; // Check if the user manually KILLED/FAILED this task-attempt... Boolean shouldFail = tasksToKill . remove ( taskid ) ; if ( shouldFail != null ) { if ( status . getRunState ( ) == TaskStatus . State . FAILED || status . getRunState ( ) == TaskStatus . State . KILLED ) { taskState = ( shouldFail ) ? TaskStatus . State . FAILED : TaskStatus . State . KILLED ; } else { taskState = ( shouldFail ) ? TaskStatus . State . FAILED_UNCLEAN : TaskStatus . State . KILLED_UNCLEAN ; } status . setRunState ( taskState ) ; addDiagnosticInfo ( taskid , "Task has been " + taskState + " by the user" ) ; } taskState = status . getRunState ( ) ; if ( taskState != TaskStatus . State . FAILED && taskState != TaskStatus . State . KILLED && taskState != TaskStatus . State . FAILED_UNCLEAN && taskState != TaskStatus . State . KILLED_UNCLEAN ) { LOG . info ( "Task '" + taskid + "' running on '" + trackerName + "' in state: '" + taskState + "' being failed!" ) ; status . setRunState ( TaskStatus . State . FAILED ) ; taskState = TaskStatus . State . FAILED ; } // tasktracker went down and failed time was not reported. if ( 0 == status . getFinishTime ( ) ) { status . setFinishTime ( JobTracker . getClock ( ) . getTime ( ) ) ; } } this . activeTasks . remove ( taskid ) ; // Since we do not fail completed reduces (whose outputs go to hdfs), we // should note this failure only for completed maps, only if this taskid; // completed this map. however if the job is done, there is no need to // manipulate completed maps if ( this . isMapTask ( ) && ! jobSetup && ! jobCleanup && isComplete ( taskid ) && jobStatus . getRunState ( ) != JobStatus . SUCCEEDED ) { this . completes -- ; // Reset the successfulTaskId since we don't have a SUCCESSFUL task now resetSuccessfulTaskid ( ) ; } // Note that there can be failures of tasks that are hosted on a machine // that has not yet registered with restarted jobtracker // recalculate the counts only if its a genuine failure if ( tasks . contains ( taskid ) ) { if ( taskState == TaskStatus . State . FAILED ) { numTaskFailures ++ ; machinesWhereFailed . add ( trackerHostName ) ; if ( maxSkipRecords > 0 ) { //skipping feature enabled LOG . debug ( "TaskInProgress adding" + status . getNextRecordRange ( ) ) ; failedRanges . add ( status . getNextRecordRange ( ) ) ; skipping = startSkipping ( ) ; } } else if ( taskState == TaskStatus . State . KILLED ) { numKilledTasks ++ ; } } if ( numTaskFailures >= maxTaskAttempts ) { LOG . info ( "TaskInProgress " + getTIPId ( ) + " has failed " + numTaskFailures + " times." ) ; kill ( ) ; } }
Indicate that one of the taskids in this TaskInProgress has failed .