HADOOP-4869. Fixes the TT-JT heartbeat to have an explicit flag for restart apart from the initialContact flag that there was earlier. Contributed by Amareshwari Sriramadasu.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@728601 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -491,6 +491,10 @@ Release 0.20.0 - Unreleased
 
     HADOOP-4889. Fix permissions in RPM packaging. (Eric Yang via cdouglas)
 
+    HADOOP-4869. Fixes the TT-JT heartbeat to have an explicit flag for 
+    restart apart from the initialContact flag that there was earlier.
+    (Amareshwari Sriramadasu via ddas)
+
 Release 0.19.1 - Unreleased
 
   IMPROVEMENTS

src/mapred/org/apache/hadoop/mapred/InterTrackerProtocol.java

@@ -30,7 +30,7 @@ interface InterTrackerProtocol extends VersionedProtocol {
   /**
    * version 3 introduced to replace 
    * emitHearbeat/pollForNewTask/pollForTaskWithClosedJob with
-   * {@link #heartbeat(TaskTrackerStatus, boolean, boolean, short)}
+   * {@link #heartbeat(TaskTrackerStatus, boolean, boolean, boolean, short)}
    * version 4 changed TaskReport for HADOOP-549.
    * version 5 introduced that removes locateMapOutputs and instead uses
    * getTaskCompletionEvents to figure finished maps and fetch the outputs
@@ -57,8 +57,10 @@ interface InterTrackerProtocol extends VersionedProtocol {
    *             (HADOOP-4035)
    * Version 22: Replaced parameter 'initialContact' with 'restarted' 
    *             in heartbeat method (HADOOP-4305) 
+   * Version 23: Added parameter 'initialContact' again in heartbeat method
+   *            (HADOOP-4869) 
    */
-  public static final long versionID = 22L;
+  public static final long versionID = 23L;
   
   public final static int TRACKERS_OK = 0;
   public final static int UNKNOWN_TASKTRACKER = 1;
@@ -77,6 +79,8 @@ interface InterTrackerProtocol extends VersionedProtocol {
    * @param status the status update
    * @param restarted <code>true</code> if the process has just started or 
    *                   restarted, <code>false</code> otherwise
+   * @param initialContact <code>true</code> if this is first interaction since
+   *                       'refresh', <code>false</code> otherwise.
    * @param acceptNewTasks <code>true</code> if the {@link TaskTracker} is
    *                       ready to accept new tasks to run.                 
    * @param responseId the last responseId successfully acted upon by the
@@ -86,6 +90,7 @@ interface InterTrackerProtocol extends VersionedProtocol {
    */
   HeartbeatResponse heartbeat(TaskTrackerStatus status, 
                               boolean restarted, 
+                              boolean initialContact,
                               boolean acceptNewTasks,
                               short responseId)
     throws IOException;

src/mapred/org/apache/hadoop/mapred/JobTracker.java

@@ -2197,11 +2197,13 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
    */
   public synchronized HeartbeatResponse heartbeat(TaskTrackerStatus status, 
                                                   boolean restarted,
+                                                  boolean initialContact,
                                                   boolean acceptNewTasks, 
                                                   short responseId) 
     throws IOException {
     LOG.debug("Got heartbeat from: " + status.getTrackerName() + 
               " (restarted: " + restarted + 
+              " initialContact: " + initialContact + 
               " acceptNewTasks: " + acceptNewTasks + ")" +
               " with responseId: " + responseId);
 
@@ -2225,7 +2227,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
       trackerToHeartbeatResponseMap.get(trackerName);
     boolean addRestartInfo = false;
 
-    if (restarted != true) {
+    if (initialContact != true) {
       // If this isn't the 'initial contact' from the tasktracker,
       // there is something seriously wrong if the JobTracker has
       // no record of the 'previous heartbeat'; if so, ask the 
@@ -2263,7 +2265,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     // Process this heartbeat 
     short newResponseId = (short)(responseId + 1);
     status.setLastSeen(now);
-    if (!processHeartbeat(status, restarted)) {
+    if (!processHeartbeat(status, initialContact)) {
+      if (prevHeartbeatResponse != null) {
+        trackerToHeartbeatResponseMap.remove(trackerName);
+      }
       return new HeartbeatResponse(newResponseId, 
                    new TaskTrackerAction[] {new ReinitTrackerAction()});
     }
@@ -2426,14 +2431,14 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
    */
   private synchronized boolean processHeartbeat(
                                  TaskTrackerStatus trackerStatus, 
-                                 boolean restarted) {
+                                 boolean initialContact) {
     String trackerName = trackerStatus.getTrackerName();
 
     synchronized (taskTrackers) {
       synchronized (trackerExpiryQueue) {
         boolean seenBefore = updateTaskTrackerStatus(trackerName,
                                                      trackerStatus);
-        if (restarted) {
+        if (initialContact) {
           // If it's first contact, then clear out 
           // any state hanging around
           if (seenBefore) {
@@ -2443,17 +2448,17 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
           // If not first contact, there should be some record of the tracker
           if (!seenBefore) {
             LOG.warn("Status from unknown Tracker : " + trackerName);
-            // This is lost tracker that came back now, if it blacklisted
-            // increment the count of blacklisted trackers in the cluster
-            if (isBlacklisted(trackerName)) {
-              faultyTrackers.numBlacklistedTrackers += 1;
-            }
-            addNewTracker(trackerStatus);
+            updateTaskTrackerStatus(trackerName, null);
             return false;
           }
         }
 
-        if (restarted) {
+        if (initialContact) {
+          // if this is lost tracker that came back now, and if it blacklisted
+          // increment the count of blacklisted trackers in the cluster
+          if (isBlacklisted(trackerName)) {
+            faultyTrackers.numBlacklistedTrackers += 1;
+          }
           addNewTracker(trackerStatus);
         }
       }

src/mapred/org/apache/hadoop/mapred/TaskTracker.java

@@ -162,6 +162,7 @@ public class TaskTracker
   volatile int mapTotal = 0;
   volatile int reduceTotal = 0;
   boolean justStarted = true;
+  boolean justInited = true;
   // Mark reduce tasks that are shuffling to rollback their events index
   Set<TaskAttemptID> shouldReset = new HashSet<TaskAttemptID>();
     
@@ -524,6 +525,7 @@ public class TaskTracker
       RPC.waitForProxy(InterTrackerProtocol.class,
                        InterTrackerProtocol.versionID, 
                        jobTrackAddr, this.fConf);
+    this.justInited = true;
     this.running = true;    
     // start the thread that will fetch map task completion events
     this.mapEventsFetcher = new MapEventsFetcherThread();
@@ -1020,7 +1022,7 @@ public class TaskTracker
         // If the TaskTracker is just starting up:
         // 1. Verify the buildVersion
         // 2. Get the system directory & filesystem
-        if(justStarted){
+        if(justInited) {
           String jobTrackerBV = jobClient.getBuildVersion();
           if(!VersionInfo.getBuildVersion().equals(jobTrackerBV)) {
             String msg = "Shutting down. Incompatible buildVersion." +
@@ -1096,6 +1098,7 @@ public class TaskTracker
         // resetting heartbeat interval from the response.
         heartbeatInterval = heartbeatResponse.getHeartbeatInterval();
         justStarted = false;
+        justInited = false;
         if (actions != null){ 
           for(TaskTrackerAction action: actions) {
             if (action instanceof LaunchTaskAction) {
@@ -1217,7 +1220,9 @@ public class TaskTracker
     // Xmit the heartbeat
     //
     HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status, 
-                                                              justStarted, askForNewTask, 
+                                                              justStarted,
+                                                              justInited,
+                                                              askForNewTask, 
                                                               heartbeatResponseId);
       
     //