private void updateTaskStatus ( TaskTrackerInfo info , TaskStatus report ) { TaskAttemptID taskId = report . getTaskID ( ) ; // Here we want strict job id comparison. if ( ! this . jobId . equals ( taskId . getJobID ( ) ) ) { LOG . warn ( "Task " + taskId + " belongs to unknown job " + taskId . getJobID ( ) ) ; return ; } TaskInProgress tip = taskLookupTable . getTIP ( taskId ) ; if ( tip == null ) { return ; } TaskStatus status = tip . getTaskStatus ( taskId ) ; TaskStatus . State knownState = ( status == null ) ? null : status . getRunState ( ) ; // Remove it from the expired task list if ( report . getRunState ( ) != TaskStatus . State . UNASSIGNED ) { expireTasks . removeTask ( taskId ) ; } // Fallback heartbeats may claim that task is RUNNING, while it was killed if ( report . getRunState ( ) == TaskStatus . State . RUNNING && ! TaskStatus . TERMINATING_STATES . contains ( knownState ) ) { expireTasks . updateTask ( taskId ) ; } // Clone TaskStatus object here, because CoronaJobInProgress // or TaskInProgress can modify this object and // the changes should not get reflected in TaskTrackerStatus. // An old TaskTrackerStatus is used later in countMapTasks, etc. job . updateTaskStatus ( tip , ( TaskStatus ) report . clone ( ) , info ) ; }
Updates job and tasks state according to TaskStatus from given TaskTracker . This function only updates internal job state it shall NOT issue any actions directly .