protected HeartbeatResponse transmitHeartBeat(
      InterTrackerProtocol jobClient, short heartbeatResponseId,
      TaskTrackerStatus status) throws IOException {
    //
    // Check if we should ask for a new Task
    //
    boolean askForNewTask;
    long localMinSpaceStart;
    synchronized (this) {
      askForNewTask =
        ((status.countOccupiedMapSlots() < maxMapSlots ||
          status.countOccupiedReduceSlots() < maxReduceSlots) &&
         acceptNewTasks);
      localMinSpaceStart = minSpaceStart;
    }
    if (askForNewTask) {
      checkLocalDirs(getLocalDirsFromConf(fConf));
      askForNewTask = enoughFreeSpace(localMinSpaceStart);
      gatherResourceStatus(status);
    }
    //add node health information

    TaskTrackerHealthStatus healthStatus = status.getHealthStatus();
    synchronized (this) {
      if (healthChecker != null) {
        healthChecker.setHealthStatus(healthStatus);
      } else {
        healthStatus.setNodeHealthy(true);
        healthStatus.setLastReported(0L);
        healthStatus.setHealthReport("");
      }
    }
    //
    // Xmit the heartbeat
    //
    HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,
                                                              justStarted,
                                                              justInited,
                                                              askForNewTask,
                                                              heartbeatResponseId);

    synchronized (this) {
      for (TaskStatus taskStatus : status.getTaskReports()) {
        if (taskStatus.getRunState() != TaskStatus.State.RUNNING &&
            taskStatus.getRunState() != TaskStatus.State.UNASSIGNED &&
            taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING &&
            !taskStatus.inTaskCleanupPhase()) {
          if (taskStatus.getIsMap()) {
            mapTotal--;
          } else {
            reduceTotal--;
          }
          try {
            myInstrumentation.completeTask(taskStatus.getTaskID());
          } catch (MetricsException me) {
            LOG.warn("Caught: " + StringUtils.stringifyException(me));
          }
          removeRunningTask(taskStatus.getTaskID());
          
          //
          // When the task attempt has entered the finished state
          // we log the counters to task log for future use
          // load the counters into Scuba, Scriber and Hive
          //
          if (fConf.getBoolean(LOG_FINISHED_TASK_COUNTERS, true)) {
            // for log format, 0 means json, else means name and value pair
            String logHeader = "TaskCountersLogged " + taskStatus.getTaskID() + " " +
              taskStatus.getFinishTime()/1000 + " ";
            if (fConf.getInt(FINISHED_TASK_COUNTERS_LOG_FORMAT, 0) == 0) {
              LOG.warn(
                  logHeader + taskStatus.getCounters().makeJsonString());
            } else {
              LOG.warn(
                  logHeader + taskStatus.getCounters().makeCompactString());
            }
          }
        }
      }

      // Clear transient status information which should only
      // be sent once to the JobTracker
      for (TaskInProgress tip: runningTasks.values()) {
        tip.getStatus().clearStatus();
      }
    }

    return heartbeatResponse;
  }
protected HeartbeatResponse transmitHeartBeat ( InterTrackerProtocol jobClient , short heartbeatResponseId , TaskTrackerStatus status ) throws IOException { boolean askForNewTask ; long localMinSpaceStart ; synchronized ( this ) { askForNewTask = ( ( status . countOccupiedMapSlots ( ) < maxMapSlots || status . countOccupiedReduceSlots ( ) < maxReduceSlots ) && acceptNewTasks ) ; localMinSpaceStart = minSpaceStart ; } if ( askForNewTask ) { checkLocalDirs ( getLocalDirsFromConf ( fConf ) ) ; askForNewTask = enoughFreeSpace ( localMinSpaceStart ) ; gatherResourceStatus ( status ) ; } TaskTrackerHealthStatus healthStatus = status . getHealthStatus ( ) ; synchronized ( this ) { if ( healthChecker != null ) { healthChecker . setHealthStatus ( healthStatus ) ; } else { healthStatus . setNodeHealthy ( true ) ; healthStatus . setLastReported ( 0L ) ; healthStatus . setHealthReport ( "" ) ; } } HeartbeatResponse heartbeatResponse = jobClient . heartbeat ( status , justStarted , justInited , askForNewTask , heartbeatResponseId ) ; synchronized ( this ) { for ( TaskStatus taskStatus : status . getTaskReports ( ) ) { if ( taskStatus . getRunState ( ) != TaskStatus . State . RUNNING && taskStatus . getRunState ( ) != TaskStatus . State . UNASSIGNED && taskStatus . getRunState ( ) != TaskStatus . State . COMMIT_PENDING && ! taskStatus . inTaskCleanupPhase ( ) ) { if ( taskStatus . getIsMap ( ) ) { mapTotal -- ; } else { reduceTotal -- ; } try { myInstrumentation . completeTask ( taskStatus . getTaskID ( ) ) ; } catch ( MetricsException me ) { LOG . warn ( "Caught: " + StringUtils . stringifyException ( me ) ) ; } removeRunningTask ( taskStatus . getTaskID ( ) ) ; if ( fConf . getBoolean ( LOG_FINISHED_TASK_COUNTERS , true ) ) { String logHeader = "TaskCountersLogged " + taskStatus . getTaskID ( ) + " " + taskStatus . getFinishTime ( ) / 1000 + " " ; if ( fConf . getInt ( FINISHED_TASK_COUNTERS_LOG_FORMAT , 0 ) == 0 ) { LOG . warn ( logHeader + taskStatus . getCounters ( ) . makeJsonString ( ) ) ; } else { LOG . warn ( logHeader + taskStatus . getCounters ( ) . makeCompactString ( ) ) ; } } } } for ( TaskInProgress tip : runningTasks . values ( ) ) { tip . getStatus ( ) . clearStatus ( ) ; } } return heartbeatResponse ; }
Build and transmit the heart beat to the JobTracker
@param jobClient The jobTracker RPC handle
@param heartbeatResponseId Last heartbeat response received
@param status TaskTrackerStatus to transmit
@return false if the tracker was unknown
@throws IOException