private void heartbeatToClusterManager() throws IOException {
    CoronaConf coronaConf = new CoronaConf(fConf);
    int numCpu = coronaConf.getInt("mapred.coronatasktracker.num.cpus",
      resourceCalculatorPlugin.getNumProcessors());
    if (numCpu == ResourceCalculatorPlugin.UNAVAILABLE) {
      numCpu = 1;
    }
    LOG.info("Will report " + numCpu + " CPUs");
    int totalMemoryMB = (int) (resourceCalculatorPlugin.getPhysicalMemorySize() / 1024D / 1024);
    ComputeSpecs total = new ComputeSpecs((short)numCpu);
    total.setNetworkMBps((short)100);
    total.setMemoryMB(totalMemoryMB);
    total.setDiskGB(
       (int)(getDiskSpace(false) / 1024D / 1024 / 1024));
    String appInfo = null;
    if (getLocalHostAddress() != null) {
      appInfo = getLocalHostAddress() + ":" + actionServerAddr.getPort();
    }
    else {
      appInfo = getLocalHostname() + ":" + actionServerAddr.getPort();
    }
    Map<ResourceType, String> resourceInfos =
        new EnumMap<ResourceType, String>(ResourceType.class);
    resourceInfos.put(ResourceType.MAP, appInfo);
    resourceInfos.put(ResourceType.REDUCE, appInfo);
    resourceInfos.put(ResourceType.JOBTRACKER, appInfo);

    while (running && !shuttingDown) {
      try {
        long now = System.currentTimeMillis();
        Thread.sleep(heartbeatCMInterval);

        float cpuUsage = resourceCalculatorPlugin.getCpuUsage();
        if (cpuUsage == ResourceCalculatorPlugin.UNAVAILABLE) {
          cpuUsage = 0;
        }
        ComputeSpecs free = new ComputeSpecs((short)(numCpu * cpuUsage / 100D));
        // TODO find free network.
        free.setNetworkMBps((short)100);
        int availableMemoryMB =
            (int)(resourceCalculatorPlugin.
                getAvailablePhysicalMemorySize() / 1024D / 1024);
        free.setMemoryMB(availableMemoryMB);
        long freeDiskSpace = getDiskSpace(true);
        long freeLogDiskSpace = getLogDiskFreeSpace();
        free.setDiskGB((int)(
          Math.min(freeDiskSpace, freeLogDiskSpace) / 1024D / 1024 / 1024));
        // TT puts it's MR specific host:port tuple here
        ClusterNodeInfo node = new ClusterNodeInfo
          (this.getName(), clusterManagerCallbackServerAddr, total);
        node.setFree(free);
        node.setResourceInfos(resourceInfos);

        LOG.debug("ClusterManager heartbeat: " + node.toString());
        if (client == null) {
          initializeClusterManagerClient();
        }
        NodeHeartbeatResponse nodeHeartbeatResponse =
          client.nodeHeartbeat(node);
        if (nodeHeartbeatResponse.restartFlag) {
          LOG.fatal("Get CM notice to exit");
          System.exit(0);
        }
        clusterManagerConnectRetries = 0;
        lastCMHeartbeat = System.currentTimeMillis();

        markUnresponsiveTasks();
        killOverflowingTasks();

        //we've cleaned up, resume normal operation
        if (!acceptNewTasks && isIdle()) {
          acceptNewTasks=true;
        }
        //The check below may not be required every iteration but we are
        //erring on the side of caution here. We have seen many cases where
        //the call to jetty's getLocalPort() returns different values at
        //different times. Being a real paranoid here.
        checkJettyPort();
      } catch (InterruptedException ie) {
        LOG.info("Interrupted. Closing down.");
        return;
      } catch (DisallowedNode ex) {
        LOG.error("CM has excluded node, shutting down TT");
        shutdown();
      } catch (SafeModeException e) {
        LOG.info("Cluster Manager is in Safe Mode");
        try {
          ClusterManagerAvailabilityChecker.
            waitWhileClusterManagerInSafeMode(coronaConf);
        } catch (IOException ie) {
          LOG.error("Could not wait while Cluster Manager is in Safe Mode ",
                    ie);
        }
      } catch (TException ex) {
        if (!shuttingDown) {
          LOG.error("Error connecting to CM. " + clusterManagerConnectRetries
            + "th retry. Retry in 10 seconds.", ex);
          closeClusterManagerClient();
          if (++clusterManagerConnectRetries >= MAX_CM_CONNECT_RETRIES) {
            LOG.error("Cannot connect to CM " + clusterManagerConnectRetries +
              " times. Shutting down TT");
            shutdown();
          }
          try {
            Thread.sleep(10000L);
          } catch (InterruptedException ie) {
          }
          ClusterManagerAvailabilityChecker.
            waitWhileClusterManagerInSafeMode(coronaConf);
        }
      }
    }
  }
private void heartbeatToClusterManager ( ) throws IOException { CoronaConf coronaConf = new CoronaConf ( fConf ) ; int numCpu = coronaConf . getInt ( "mapred.coronatasktracker.num.cpus" , resourceCalculatorPlugin . getNumProcessors ( ) ) ; if ( numCpu == ResourceCalculatorPlugin . UNAVAILABLE ) { numCpu = 1 ; } LOG . info ( "Will report " + numCpu + " CPUs" ) ; int totalMemoryMB = ( int ) ( resourceCalculatorPlugin . getPhysicalMemorySize ( ) / 1024D / 1024 ) ; ComputeSpecs total = new ComputeSpecs ( ( short ) numCpu ) ; total . setNetworkMBps ( ( short ) 100 ) ; total . setMemoryMB ( totalMemoryMB ) ; total . setDiskGB ( ( int ) ( getDiskSpace ( false ) / 1024D / 1024 / 1024 ) ) ; String appInfo = null ; if ( getLocalHostAddress ( ) != null ) { appInfo = getLocalHostAddress ( ) + ":" + actionServerAddr . getPort ( ) ; } else { appInfo = getLocalHostname ( ) + ":" + actionServerAddr . getPort ( ) ; } Map < ResourceType , String > resourceInfos = new EnumMap < ResourceType , String > ( ResourceType . class ) ; resourceInfos . put ( ResourceType . MAP , appInfo ) ; resourceInfos . put ( ResourceType . REDUCE , appInfo ) ; resourceInfos . put ( ResourceType . JOBTRACKER , appInfo ) ; while ( running && ! shuttingDown ) { try { long now = System . currentTimeMillis ( ) ; Thread . sleep ( heartbeatCMInterval ) ; float cpuUsage = resourceCalculatorPlugin . getCpuUsage ( ) ; if ( cpuUsage == ResourceCalculatorPlugin . UNAVAILABLE ) { cpuUsage = 0 ; } ComputeSpecs free = new ComputeSpecs ( ( short ) ( numCpu * cpuUsage / 100D ) ) ; free . setNetworkMBps ( ( short ) 100 ) ; int availableMemoryMB = ( int ) ( resourceCalculatorPlugin . getAvailablePhysicalMemorySize ( ) / 1024D / 1024 ) ; free . setMemoryMB ( availableMemoryMB ) ; long freeDiskSpace = getDiskSpace ( true ) ; long freeLogDiskSpace = getLogDiskFreeSpace ( ) ; free . setDiskGB ( ( int ) ( Math . min ( freeDiskSpace , freeLogDiskSpace ) / 1024D / 1024 / 1024 ) ) ; ClusterNodeInfo node = new ClusterNodeInfo ( this . getName ( ) , clusterManagerCallbackServerAddr , total ) ; node . setFree ( free ) ; node . setResourceInfos ( resourceInfos ) ; LOG . debug ( "ClusterManager heartbeat: " + node . toString ( ) ) ; if ( client == null ) { initializeClusterManagerClient ( ) ; } NodeHeartbeatResponse nodeHeartbeatResponse = client . nodeHeartbeat ( node ) ; if ( nodeHeartbeatResponse . restartFlag ) { LOG . fatal ( "Get CM notice to exit" ) ; System . exit ( 0 ) ; } clusterManagerConnectRetries = 0 ; lastCMHeartbeat = System . currentTimeMillis ( ) ; markUnresponsiveTasks ( ) ; killOverflowingTasks ( ) ; if ( ! acceptNewTasks && isIdle ( ) ) { acceptNewTasks = true ; } checkJettyPort ( ) ; } catch ( InterruptedException ie ) { LOG . info ( "Interrupted. Closing down." ) ; return ; } catch ( DisallowedNode ex ) { LOG . error ( "CM has excluded node, shutting down TT" ) ; shutdown ( ) ; } catch ( SafeModeException e ) { LOG . info ( "Cluster Manager is in Safe Mode" ) ; try { ClusterManagerAvailabilityChecker . waitWhileClusterManagerInSafeMode ( coronaConf ) ; } catch ( IOException ie ) { LOG . error ( "Could not wait while Cluster Manager is in Safe Mode " , ie ) ; } } catch ( TException ex ) { if ( ! shuttingDown ) { LOG . error ( "Error connecting to CM. " + clusterManagerConnectRetries + "th retry. Retry in 10 seconds." , ex ) ; closeClusterManagerClient ( ) ; if ( ++ clusterManagerConnectRetries >= MAX_CM_CONNECT_RETRIES ) { LOG . error ( "Cannot connect to CM " + clusterManagerConnectRetries + " times. Shutting down TT" ) ; shutdown ( ) ; } try { Thread . sleep ( 10000L ) ; } catch ( InterruptedException ie ) { } ClusterManagerAvailabilityChecker . waitWhileClusterManagerInSafeMode ( coronaConf ) ; } } } }
Main service loop.  Will stay in this loop forever.