public Task addRunningTask ( TaskAttemptID taskid , String taskTracker , boolean taskCleanup ) { // 1 slot is enough for taskCleanup task int numSlotsNeeded = taskCleanup ? 1 : numSlotsRequired ; // create the task Task t = null ; if ( isMapTask ( ) ) { LOG . debug ( "attempt " + numTaskFailures + " sending skippedRecords " + failedRanges . getIndicesCount ( ) ) ; String splitClass = null ; BytesWritable split ; if ( ! jobSetup && ! jobCleanup ) { splitClass = rawSplit . getClassName ( ) ; split = rawSplit . getBytes ( ) ; } else { split = new BytesWritable ( ) ; } t = new MapTask ( jobFile , taskid , partition , splitClass , split , numSlotsNeeded , job . getUser ( ) ) ; } else { t = new ReduceTask ( jobFile , taskid , partition , numMaps , numSlotsNeeded , job . getUser ( ) ) ; } if ( jobCleanup ) { t . setJobCleanupTask ( ) ; } if ( jobSetup ) { t . setJobSetupTask ( ) ; } if ( taskCleanup ) { t . setTaskCleanupTask ( ) ; t . setState ( taskStatuses . get ( taskid ) . getRunState ( ) ) ; cleanupTasks . put ( taskid , taskTracker ) ; } t . setConf ( conf ) ; LOG . debug ( "Launching task with skipRanges:" + failedRanges . getSkipRanges ( ) ) ; t . setSkipRanges ( failedRanges . getSkipRanges ( ) ) ; t . setSkipping ( skipping ) ; if ( failedRanges . isTestAttempt ( ) ) { t . setWriteSkipRecs ( false ) ; } if ( activeTasks . size ( ) >= 1 ) { speculativeTaskId = taskid ; } else { speculativeTaskId = null ; } activeTasks . put ( taskid , taskTracker ) ; tasks . add ( taskid ) ; // Ask JobTracker to note that the task exists // jobtracker.createTaskEntry(taskid, taskTracker, this); /* // code to find call paths to createTaskEntry StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace(); boolean found = false; for (StackTraceElement s: stackTraceElements) { if (s.getMethodName().indexOf("heartbeat") != -1 || s.getMethodName().indexOf("findTask") != -1 || s.getMethodName().indexOf("createAndAddAttempt") != -1 || s.getMethodName().indexOf("processTaskAttempt") != -1) { found = true; break; } } if (!found) { RuntimeException e = new RuntimeException ("calling addRunningTask from outside heartbeat"); LOG.info(StringUtils.stringifyException(e)); throw (e); } */ // check and set the first attempt if ( firstTaskId == null ) { firstTaskId = taskid ; } return t ; }
Adds a previously running task to this tip . This is used in case of jobtracker restarts .