HADOOP-5067. Fixes TaskInProgress.java to keep track of count of failed and killed tasks correctly. Contributed by Amareshwari Sriramadasu.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.19@743974 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -102,6 +102,9 @@ Release 0.19.1 - Unreleased
     HADOOP-5166. Fix JobTracker restart to work when ACLs are configured
     for the JobTracker. (Amar Kamat via yhemanth).
 
+    HADOOP-5067. Fixes TaskInProgress.java to keep track of count of failed and
+    killed tasks correctly. (Amareshwari Sriramadasu via ddas)
+
 Release 0.19.0 - 2008-11-18
 
   INCOMPATIBLE CHANGES

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

@@ -94,6 +94,8 @@ class TaskInProgress {
   // Map from task Id -> TaskTracker Id, contains tasks that are
   // currently runnings
   private TreeMap<TaskAttemptID, String> activeTasks = new TreeMap<TaskAttemptID, String>();
+  // All attempt Ids of this TIP
+  private TreeSet<TaskAttemptID> tasks = new TreeSet<TaskAttemptID>();
   private JobConf conf;
   private Map<TaskAttemptID,List<String>> taskDiagnosticData =
     new TreeMap<TaskAttemptID,List<String>>();
@@ -585,9 +587,7 @@ class TaskInProgress {
       }
     }
 
-    // Note that there can be failures of tasks that are hosted on a machine 
-    // that has not yet registered with restarted jobtracker
-    boolean isPresent = this.activeTasks.remove(taskid) != null;
+    this.activeTasks.remove(taskid);
     
     // Since we do not fail completed reduces (whose outputs go to hdfs), we 
     // should note this failure only for completed maps, only if this taskid;
@@ -601,8 +601,10 @@ class TaskInProgress {
       resetSuccessfulTaskid();
     }
 
+    // Note that there can be failures of tasks that are hosted on a machine 
+    // that has not yet registered with restarted jobtracker
     // recalculate the counts only if its a genuine failure
-    if (isPresent) {
+    if (tasks.contains(taskid)) {
       if (taskState == TaskStatus.State.FAILED) {
         numTaskFailures++;
         machinesWhereFailed.add(trackerHostName);
@@ -907,6 +909,7 @@ class TaskInProgress {
     }
 
     activeTasks.put(taskid, taskTracker);
+    tasks.add(taskid);
 
     // Ask JobTracker to note that the task exists
     jobtracker.createTaskEntry(taskid, taskTracker, this);

src/test/org/apache/hadoop/mapred/MiniMRCluster.java

@@ -235,6 +235,9 @@ public class MiniMRCluster {
     return jobTracker;
   }
   
+  TaskTrackerRunner getTaskTrackerRunner(int id) {
+    return taskTrackerList.get(id);
+  }
   /**
    * Get the number of task trackers in the cluster
    */

src/test/org/apache/hadoop/mapred/TestLostTracker.java

@@ -0,0 +1,158 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.mapred.TestJobTrackerRestart;
+
+import junit.framework.TestCase;
+import java.io.*;
+
+public class TestLostTracker extends TestCase {
+  final Path testDir = new Path("/jt-lost-tt");
+  final Path inDir = new Path(testDir, "input");
+  final Path shareDir = new Path(testDir, "share");
+  final Path outputDir = new Path(testDir, "output");
+  
+  private JobConf configureJob(JobConf conf, int[] maps, int[] reduces,
+                               String mapSignal, String redSignal) 
+  throws IOException {
+    JobPriority[] priority = new JobPriority[] {JobPriority.NORMAL};
+    return TestJobTrackerRestart.getJobs(conf, priority, 
+                                         maps, reduces, outputDir, inDir, 
+                                         mapSignal, redSignal)[0];
+  }
+  
+  public void testLostTracker(MiniDFSCluster dfs,
+                              MiniMRCluster mr) 
+  throws IOException {
+    FileSystem fileSys = dfs.getFileSystem();
+    JobConf jobConf = mr.createJobConf();
+    int numMaps = 10;
+    int numReds = 1;
+    String mapSignalFile = TestJobTrackerRestart.getMapSignalFile(shareDir);
+    String redSignalFile = TestJobTrackerRestart.getReduceSignalFile(shareDir);
+    
+    // Configure the job
+    JobConf job = configureJob(jobConf, new int[] {numMaps}, 
+                               new int[] {numReds}, 
+                               mapSignalFile, redSignalFile);
+      
+    TestJobTrackerRestart.cleanUp(fileSys, shareDir);
+    
+    // Submit the job   
+    JobClient jobClient = new JobClient(job);
+    RunningJob rJob = jobClient.submitJob(job);
+    JobID id = rJob.getID();
+    
+    // wait for the job to be inited
+    mr.initializeJob(id);
+    
+    // Make sure that the master job is 50% completed
+    while (TestJobTrackerRestart.getJobStatus(jobClient, id).mapProgress() 
+           < 0.5f) {
+      TestJobTrackerRestart.waitFor(10);
+    }
+    // get a completed task on 1st tracker 
+    TaskAttemptID taskid = mr.getTaskTrackerRunner(0).getTaskTracker().
+                              getNonRunningTasks().get(0).getTaskID();
+
+    // Kill the 1st tasktracker
+    mr.stopTaskTracker(0);
+    
+    // Signal all the maps to complete
+    TestJobTrackerRestart.signalTasks(dfs, fileSys, true, 
+                                      mapSignalFile, redSignalFile);
+    
+    // Signal the reducers to complete
+    TestJobTrackerRestart.signalTasks(dfs, fileSys, false, 
+                                      mapSignalFile, redSignalFile);
+    // wait for the job to complete
+    TestJobTrackerRestart.waitTillDone(jobClient);
+
+    // Check if the tasks on the lost tracker got killed and re-executed
+    assertTrue(jobClient.getClusterStatus().getTaskTrackers() 
+                < mr.getNumTaskTrackers());
+    assertEquals(JobStatus.SUCCEEDED, rJob.getJobState());
+    TaskInProgress tip = mr.getJobTrackerRunner().getJobTracker().
+                         getTip(taskid.getTaskID());
+    assertTrue(tip.isComplete());
+    assertEquals(tip.numKilledTasks(), 1);
+  }
+  
+  public void testLostTracker() throws IOException {
+    String namenode = null;
+    MiniDFSCluster dfs = null;
+    MiniMRCluster mr = null;
+    FileSystem fileSys = null;
+
+    try {
+      Configuration conf = new Configuration();
+      conf.setBoolean("dfs.replication.considerLoad", false);
+      dfs = new MiniDFSCluster(conf, 1, true, null, null);
+      dfs.waitActive();
+      fileSys = dfs.getFileSystem();
+      
+      // clean up
+      fileSys.delete(testDir, true);
+      
+      if (!fileSys.mkdirs(inDir)) {
+        throw new IOException("Mkdirs failed to create " + inDir.toString());
+      }
+
+      // Write the input file
+      TestRackAwareTaskPlacement.writeFile(dfs.getNameNode(), conf, 
+                                           new Path(inDir + "/file"), 
+                                           (short)1);
+
+      dfs.startDataNodes(conf, 1, true, null, null, null, null);
+      dfs.waitActive();
+
+      namenode = (dfs.getFileSystem()).getUri().getHost() + ":" 
+                 + (dfs.getFileSystem()).getUri().getPort();
+
+      JobConf jtConf = new JobConf();
+      jtConf.setInt("mapred.tasktracker.map.tasks.maximum", 1);
+      jtConf.setInt("mapred.tasktracker.reduce.tasks.maximum", 1);
+      jtConf.setLong("mapred.tasktracker.expiry.interval", 10 * 1000);
+      jtConf.setInt("mapred.reduce.copy.backoff", 4);
+      
+      mr = new MiniMRCluster(2, namenode, 1, null, null, jtConf);
+      
+      // Test Lost tracker case
+      testLostTracker(dfs, mr);
+    } finally {
+      if (mr != null) {
+        try {
+          mr.shutdown();
+        } catch (Exception e) {}
+      }
+      if (dfs != null) {
+        try {
+          dfs.shutdown();
+        } catch (Exception e) {}
+      }
+    }
+  }
+
+  public static void main(String[] args) throws IOException {
+    new TestLostTracker().testLostTracker();
+  }
+}