HADOOP-1874. Move task-outputs' promotion/discard to a separate thread distinct from the main heartbeat-processing thread. The main upside being that we do not lock-up the JobTracker during HDFS operations, which otherwise may lead to lost tasktrackers if the NameNode is unresponsive. Contributed by Devaraj Das.

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

CHANGES.txt

@@ -253,6 +253,12 @@ Trunk (unreleased changes)
     HADOOP-1992.  Fix the performance degradation in the sort validator. 
     (acmurthy via omalley)
 
+    HADOOP-1874.  Move task-outputs' promotion/discard to a separate thread
+    distinct from the main heartbeat-processing thread. The main upside being 
+    that we do not lock-up the JobTracker during HDFS operations, which
+    otherwise may lead to lost tasktrackers if the NameNode is unresponsive.
+    (Devaraj Das via acmurthy)
+
   IMPROVEMENTS
 
     HADOOP-1908. Restructure data node code so that block sending and 

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

@@ -37,7 +37,6 @@ import org.apache.hadoop.mapred.JobTracker.JobTrackerMetrics;
 import org.apache.hadoop.metrics.MetricsContext;
 import org.apache.hadoop.metrics.MetricsRecord;
 import org.apache.hadoop.metrics.MetricsUtil;
-import org.apache.hadoop.util.StringUtils;
 
 /*************************************************************
  * JobInProgress maintains all the info for keeping
@@ -380,6 +379,16 @@ class JobInProgress {
       TaskTrackerStatus ttStatus = 
         this.jobtracker.getTaskTracker(status.getTaskTracker());
       String httpTaskLogLocation = null; 
+
+      if (state == TaskStatus.State.COMMIT_PENDING ||
+          state == TaskStatus.State.FAILED ||
+          state == TaskStatus.State.KILLED) {
+        JobWithTaskContext j = new JobWithTaskContext(this, tip, 
+                                                      status.getTaskId(),
+                                                      metrics);
+        jobtracker.addToCommitQueue(j);
+      }
+
       if (null != ttStatus){
         httpTaskLogLocation = "http://" + ttStatus.getHost() + ":" + 
           ttStatus.getHttpPort() + "/tasklog?plaintext=true&taskid=" +
@@ -388,7 +397,7 @@ class JobInProgress {
 
       TaskCompletionEvent taskEvent = null;
       if (state == TaskStatus.State.SUCCEEDED) {
-        boolean complete = false;
+        completedTask(tip, status, metrics);
         taskEvent = new TaskCompletionEvent(
                                             taskCompletionEventTracker, 
                                             status.getTaskId(),
@@ -397,28 +406,14 @@ class JobInProgress {
                                             TaskCompletionEvent.Status.SUCCEEDED,
                                             httpTaskLogLocation 
                                            );
-        try {
-          complete = completedTask(tip, status, metrics);
-        } catch (IOException ioe) {
-          // Oops! Failed to copy the task's output to its final place;
-          // fail the task!
-          failedTask(tip, status.getTaskId(), 
-                     "Failed to copy reduce's output", 
-                     (tip.isMapTask() ? 
-                         TaskStatus.Phase.MAP : 
-                         TaskStatus.Phase.REDUCE), 
-                     TaskStatus.State.FAILED, 
-                     status.getTaskTracker(), null);
-          LOG.info("Failed to copy the output of " + status.getTaskId() + 
-                   " with: " + StringUtils.stringifyException(ioe));
-          return;
-        }
-        
-        if (complete) {
-          tip.setSuccessEventNumber(taskCompletionEventTracker);
-        } else {
-          taskEvent.setTaskStatus(TaskCompletionEvent.Status.KILLED);
-        }
+        tip.setSuccessEventNumber(taskCompletionEventTracker); 
+      }
+      //For a failed task update the JT datastructures.For the task state where
+      //only the COMMIT is pending, delegate everything to the JT thread. For
+      //failed tasks we want the JT to schedule a reexecution ASAP (and not go
+      //via the queue for the datastructures' updates).
+      else if (state == TaskStatus.State.COMMIT_PENDING) {
+        return;
       } else if (state == TaskStatus.State.FAILED ||
                  state == TaskStatus.State.KILLED) {
         // Get the event number for the (possibly) previously successful
@@ -771,7 +766,7 @@ class JobInProgress {
   public synchronized boolean completedTask(TaskInProgress tip, 
                                          TaskStatus status,
                                          JobTrackerMetrics metrics) 
-  throws IOException {
+  {
     String taskid = status.getTaskId();
         
     // Sanity check: is the TIP already complete? 
@@ -928,14 +923,10 @@ class JobInProgress {
                           TaskStatus status, String trackerName,
                           boolean wasRunning, boolean wasComplete,
                           JobTrackerMetrics metrics) {
-    if(status.getRunState() == TaskStatus.State.KILLED ) {
-      tip.taskKilled(taskid, trackerName, this.status);
-    }
-    else {
-      // Mark the taskid as a 'failure'
-      tip.incompleteSubTask(taskid, trackerName, this.status);
-    }
-
+    
+    // Mark the taskid as FAILED or KILLED
+    tip.incompleteSubTask(taskid, trackerName, this.status);
+   
     boolean isRunning = tip.isRunning();
     boolean isComplete = tip.isComplete();
         
@@ -1065,7 +1056,7 @@ class JobInProgress {
                                                     reason,
                                                     reason,
                                                     trackerName, phase,
-                                                    tip.getCounters());
+                                                    null);
     updateTaskStatus(tip, status, metrics);
     JobHistory.Task.logFailed(profile.getJobId(), tip.getTIPId(), 
                               tip.isMapTask() ? Values.MAP.name() : Values.REDUCE.name(), 
@@ -1178,4 +1169,30 @@ class JobInProgress {
       mapTaskIdToFetchFailuresMap.remove(mapTaskId);
     }
   }
+  
+  static class JobWithTaskContext {
+    private JobInProgress job;
+    private TaskInProgress tip;
+    private String taskId;
+    private JobTrackerMetrics metrics;
+    JobWithTaskContext(JobInProgress job, TaskInProgress tip, 
+        String taskId, JobTrackerMetrics metrics) {
+      this.job = job;
+      this.tip = tip;
+      this.taskId = taskId;
+      this.metrics = metrics;
+    }
+    JobInProgress getJob() {
+      return job;
+    }
+    TaskInProgress getTIP() {
+      return tip;
+    }
+    String getTaskId() {
+      return taskId;
+    }
+    JobTrackerMetrics getJobTrackerMetrics() {
+      return metrics;
+    }
+  }
 }

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

@@ -39,6 +39,7 @@ import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.Vector;
+import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -609,6 +610,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
   Path systemDir = null;
   private JobConf conf;
 
+  private Thread taskCommitThread;
+  
   /**
    * Start the JobTracker process, listen on the indicated port
    */
@@ -663,15 +666,6 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
 
     myMetrics = new JobTrackerMetrics(this, jobConf);
     
-    this.expireTrackersThread = new Thread(this.expireTrackers,
-                                           "expireTrackers");
-    this.expireTrackersThread.start();
-    this.retireJobsThread = new Thread(this.retireJobs, "retireJobs");
-    this.retireJobsThread.start();
-    this.initJobsThread = new Thread(this.initJobs, "initJobs");
-    this.initJobsThread.start();
-    expireLaunchingTaskThread.start();
-        
     // The rpc/web-server ports can be ephemeral ports... 
     // ... ensure we have the correct info
     this.port = interTrackerServer.getListenerAddress().getPort();
@@ -726,6 +720,17 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
    * Run forever
    */
   public void offerService() throws InterruptedException {
+    this.expireTrackersThread = new Thread(this.expireTrackers,
+                                          "expireTrackers");
+    this.expireTrackersThread.start();
+    this.retireJobsThread = new Thread(this.retireJobs, "retireJobs");
+    this.retireJobsThread.start();
+    this.initJobsThread = new Thread(this.initJobs, "initJobs");
+    this.initJobsThread.start();
+    expireLaunchingTaskThread.start();
+    this.taskCommitThread = new TaskCommitQueue();
+    this.taskCommitThread.start();
+
     this.interTrackerServer.join();
     LOG.info("Stopped interTrackerServer");
   }
@@ -781,6 +786,16 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
         ex.printStackTrace();
       }
     }
+    if (this.taskCommitThread != null) {
+      LOG.info("Stopping TaskCommit thread");
+      this.taskCommitThread.interrupt();
+      try {
+        this.taskCommitThread.interrupt();
+        this.taskCommitThread.join();
+      } catch (InterruptedException ex) {
+        ex.printStackTrace();
+      }
+    }
     LOG.info("stopped all jobtracker services");
     return;
   }
@@ -853,7 +868,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
   void markCompletedJob(JobInProgress job) {
     for (TaskInProgress tip : job.getMapTasks()) {
       for (TaskStatus taskStatus : tip.getTaskStatuses()) {
-        if (taskStatus.getRunState() != TaskStatus.State.RUNNING) {
+        if (taskStatus.getRunState() != TaskStatus.State.RUNNING && 
+            taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING) {
           markCompletedTaskAttempt(taskStatus.getTaskTracker(), 
                                    taskStatus.getTaskId());
         }
@@ -861,7 +877,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     }
     for (TaskInProgress tip : job.getReduceTasks()) {
       for (TaskStatus taskStatus : tip.getTaskStatuses()) {
-        if (taskStatus.getRunState() != TaskStatus.State.RUNNING) {
+        if (taskStatus.getRunState() != TaskStatus.State.RUNNING &&
+            taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING) {
           markCompletedTaskAttempt(taskStatus.getTaskTracker(), 
                                    taskStatus.getTaskId());
         }
@@ -1837,6 +1854,146 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     }
   }
 
+  public void addToCommitQueue(JobInProgress.JobWithTaskContext j) {
+    ((TaskCommitQueue)taskCommitThread).addToQueue(j);
+  }
+  //This thread takes care of things like moving outputs to their final
+  //locations & deleting temporary outputs
+  private class TaskCommitQueue extends Thread {
+    
+    private LinkedBlockingQueue<JobInProgress.JobWithTaskContext> queue = 
+            new LinkedBlockingQueue <JobInProgress.JobWithTaskContext>();
+        
+    public TaskCommitQueue() {
+      setName("Task Commit Thread");
+      setDaemon(true);
+    }
+    
+    public void addToQueue(JobInProgress.JobWithTaskContext j) {
+      while (!queue.add(j)) {
+        LOG.warn("Couldn't add to the Task Commit queue now. Will " +
+                 "try again");
+        try {
+          Thread.sleep(2000);
+        } catch (InterruptedException ie) {}
+      }
+    }
+       
+    public void run() {
+      while (!isInterrupted()) {
+        JobInProgress.JobWithTaskContext j;
+        try {
+          j = queue.take();
+        } catch (InterruptedException ie) {
+          return;
+        }
+        JobInProgress job = j.getJob();
+        TaskInProgress tip = j.getTIP();
+        String taskid = j.getTaskId();
+        JobTrackerMetrics metrics = j.getJobTrackerMetrics();
+        Task t;
+        TaskStatus status;
+        boolean isTipComplete = false;
+        TaskStatus.State state;
+        synchronized (JobTracker.this) {
+          synchronized (job) {
+            synchronized (tip) {
+              status = tip.getTaskStatus(taskid);
+              t = tip.getTaskObject(taskid);
+              state = status.getRunState();
+              isTipComplete = tip.isComplete();
+            }
+          }
+        }
+        try {
+          //For COMMIT_PENDING tasks, we save the task output in the dfs
+          //as well as manipulate the JT datastructures to reflect a
+          //successful task. This guarantees that we don't declare a task
+          //as having succeeded until we have successfully completed the
+          //dfs operations.
+          //For failed tasks, we just do the dfs operations here. The
+          //datastructures updates is done earlier as soon as the failure
+          //is detected so that the JT can immediately schedule another
+          //attempt for that task.
+          if (state == TaskStatus.State.COMMIT_PENDING) {
+            if (!isTipComplete) {
+              t.saveTaskOutput();
+            }
+            synchronized (JobTracker.this) {
+              //do a check for the case where after the task went to
+              //COMMIT_PENDING, it was lost. So although we would have
+              //saved the task output, we cannot declare it a SUCCESS.
+              TaskStatus newStatus = null;
+              synchronized (job) {
+                synchronized (tip) {
+                  status = tip.getTaskStatus(taskid);
+                  if (!isTipComplete) {
+                    if (status.getRunState() != 
+                         TaskStatus.State.COMMIT_PENDING) {
+                      state = TaskStatus.State.KILLED;
+                    } else {
+                      state = TaskStatus.State.SUCCEEDED;
+                    }
+                  } else {
+                    tip.addDiagnosticInfo(t.getTaskId(),"Already completed " +
+                                                        "TIP");
+                    state = TaskStatus.State.KILLED;
+
+                  }
+                  //create new status if required. If the state changed from
+                  //COMMIT_PENDING to KILLED in the JobTracker, while we were
+                  //saving the output,the JT would have called updateTaskStatus
+                  //and we don't need to call it again
+                  if (status.getRunState() == TaskStatus.State.COMMIT_PENDING){
+                    newStatus = TaskStatus.createTaskStatus(
+                        tip.isMapTask(), 
+                        taskid,
+                        state == TaskStatus.State.SUCCEEDED ? 1.0f : 0.0f,
+                            state,
+                            status.getDiagnosticInfo(),
+                            status.getStateString(),
+                            status.getTaskTracker(), status.getPhase(),
+                            status.getCounters());
+                  }
+                }
+                if (newStatus != null) {
+                  job.updateTaskStatus(tip, newStatus, metrics);
+                }
+              }
+            }
+          }
+        } catch (IOException ioe) {
+          // Oops! Failed to copy the task's output to its final place;
+          // fail the task!
+          state = TaskStatus.State.FAILED;
+          synchronized (JobTracker.this) {
+            job.failedTask(tip, status.getTaskId(), 
+                     "Failed to rename output with the exception: " + 
+                     StringUtils.stringifyException(ioe), 
+                     (tip.isMapTask() ? 
+                         TaskStatus.Phase.MAP : 
+                         TaskStatus.Phase.REDUCE), 
+                     TaskStatus.State.FAILED,  
+                         status.getTaskTracker(), null);
+          }
+          LOG.info("Failed to rename the output of " + status.getTaskId() + 
+                   " with: " + StringUtils.stringifyException(ioe));
+        }
+        if (state == TaskStatus.State.FAILED || 
+            state == TaskStatus.State.KILLED) {
+          try {
+            t.discardTaskOutput();
+          } catch (IOException ioe) { 
+            LOG.info("Failed to discard the output of task " + 
+                status.getTaskId() + " with: " + 
+                StringUtils.stringifyException(ioe));
+          }
+        }
+      }
+    }
+  }
+  
+
   /**
    * Get the localized job file path on the job trackers local file system
    * @param jobId id of the job

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

@@ -438,21 +438,32 @@ abstract class Task implements Writable, Configurable {
   
   private Path getFinalPath(Path jobOutputDir, Path taskOutput) {
     URI relativePath = taskOutputPath.toUri().relativize(taskOutput.toUri());
-    return new Path(jobOutputDir, relativePath.getPath());
+    if (relativePath.getPath().length() > 0) {
+      return new Path(jobOutputDir, relativePath.getPath());
+    } else {
+      return jobOutputDir;
+    }
   }
   
   private void moveTaskOutputs(FileSystem fs, Path jobOutputDir, Path taskOutput) 
   throws IOException {
     if (fs.isFile(taskOutput)) {
       Path finalOutputPath = getFinalPath(jobOutputDir, taskOutput);
-      fs.mkdirs(finalOutputPath.getParent());
       if (!fs.rename(taskOutput, finalOutputPath)) {
-        throw new IOException("Failed to save output of task: " + 
-                getTaskId());
+        if (!fs.delete(finalOutputPath)) {
+          throw new IOException("Failed to delete earlier output of task: " + 
+                  getTaskId());
+        }
+        if (!fs.rename(taskOutput, finalOutputPath)) {
+          throw new IOException("Failed to save output of task: " + 
+                  getTaskId());
+        }
       }
       LOG.debug("Moved " + taskOutput + " to " + finalOutputPath);
     } else if(fs.isDirectory(taskOutput)) {
       Path[] paths = fs.listPaths(taskOutput);
+      Path finalOutputPath = getFinalPath(jobOutputDir, taskOutput);
+      fs.mkdirs(finalOutputPath);
       if (paths != null) {
         for (Path path : paths) {
           moveTaskOutputs(fs, jobOutputDir, path);

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

@@ -104,7 +104,6 @@ class TaskInProgress {
 
   // Map from taskId -> Task
   private Map<String, Task> tasks = new TreeMap<String, Task>();
-  boolean savedTaskOutput = false;
 
   private TreeSet<String> machinesWhereFailed = new TreeSet<String>();
   private TreeSet<String> tasksReportedClosed = new TreeSet<String>();
@@ -185,6 +184,15 @@ class TaskInProgress {
     return partition;
   }    
 
+  public boolean isOnlyCommitPending() {
+    for (TaskStatus t : taskStatuses.values()) {
+      if (t.getRunState() == TaskStatus.State.COMMIT_PENDING) {
+        return true;
+      }
+    }
+    return false;
+  }
+  
   /**
    * Initialization common to Map and Reduce
    */
@@ -218,6 +226,15 @@ class TaskInProgress {
     return split != null;
   }
     
+  /**
+   * Return the Task object associated with a taskId
+   * @param taskId
+   * @return
+   */  
+  public Task getTaskObject(String taskId) {
+    return tasks.get(taskId);
+  }
+  
   /**
    * Is this tip currently running any tasks?
    * @return true if any tasks are running
@@ -231,7 +248,7 @@ class TaskInProgress {
    * 
    * @return <code>true</code> if the tip is complete, else <code>false</code>
    */
-  public boolean isComplete() {
+  public synchronized boolean isComplete() {
     return (completes > 0);
   }
 
@@ -350,7 +367,7 @@ class TaskInProgress {
    * @param taskId id of the task 
    * @param diagInfo diagnostic information for the task
    */
-  private void addDiagnosticInfo(String taskId, String diagInfo) {
+  public void addDiagnosticInfo(String taskId, String diagInfo) {
     List<String> diagHistory = taskDiagnosticData.get(taskId);
     if (diagHistory == null) {
       diagHistory = new ArrayList<String>();
@@ -396,7 +413,8 @@ class TaskInProgress {
       if (newState == TaskStatus.State.RUNNING &&
           (oldState == TaskStatus.State.FAILED || 
            oldState == TaskStatus.State.KILLED || 
-           oldState == TaskStatus.State.SUCCEEDED)) {
+           oldState == TaskStatus.State.SUCCEEDED ||
+           oldState == TaskStatus.State.COMMIT_PENDING)) {
         return false;
       }
           
@@ -419,10 +437,18 @@ class TaskInProgress {
     //
     // Note the failure and its location
     //
-    LOG.info("Task '" + taskid + "' has been lost.");
     TaskStatus status = taskStatuses.get(taskid);
     TaskStatus.State taskState = TaskStatus.State.FAILED;
     if (status != null) {
+      // Check if the user manually KILLED/FAILED this task-attempt...
+      Boolean shouldFail = tasksToKill.remove(taskid);
+      if (shouldFail != null) {
+        taskState = (shouldFail) ? TaskStatus.State.FAILED :
+                                   TaskStatus.State.KILLED;
+        status.setRunState(taskState);
+        addDiagnosticInfo(taskid, "Task has been " + taskState + " by the user" );
+      }
+ 
       taskState = status.getRunState();
       if (taskState != TaskStatus.State.FAILED && 
               taskState != TaskStatus.State.KILLED) {
@@ -441,24 +467,18 @@ class TaskInProgress {
     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; however if the job
-    // is done, there is no need to manipulate completed maps
-    if (this.completes > 0 && this.isMapTask() && 
+    // should note this failure only for completed maps, only if this taskid;
+    // completed this map. however if the job is done, there is no need to 
+    // manipulate completed maps
+    if (this.isMapTask() && isComplete(taskid) && 
         jobStatus.getRunState() != JobStatus.SUCCEEDED) {
       this.completes--;
     }
 
-    // Discard task output
-    Task t = tasks.get(taskid);
-    try {
-      t.discardTaskOutput();
-    } catch (IOException ioe) {
-      LOG.info("Failed to discard output of task '" + taskid + "' with " + 
-              StringUtils.stringifyException(ioe));
-    }
 
     if (taskState == TaskStatus.State.FAILED) {
       numTaskFailures++;
+      machinesWhereFailed.add(trackerName);
     } else {
       numKilledTasks++;
     }
@@ -467,7 +487,6 @@ class TaskInProgress {
       LOG.info("TaskInProgress " + getTIPId() + " has failed " + numTaskFailures + " times.");
       kill();
     }
-    machinesWhereFailed.add(trackerName);
   }
 
   /**
@@ -490,14 +509,6 @@ class TaskInProgress {
    * taskid as {@link TaskStatus.State.KILLED}. 
    */
   void alreadyCompletedTask(String taskid) {
-    Task t = tasks.get(taskid);
-    try {
-      t.discardTaskOutput();
-    } catch (IOException ioe) {
-      LOG.info("Failed to discard output of task '" + taskid + "' with " + 
-              StringUtils.stringifyException(ioe));
-    }
-
     // 'KILL' the task 
     completedTask(taskid, TaskStatus.State.KILLED);
     
@@ -512,29 +523,11 @@ class TaskInProgress {
    * Indicate that one of the taskids in this TaskInProgress
    * has successfully completed!
    */
-  public void completed(String taskid) throws IOException {
-    //
-    // Finalize the task's output
-    //
-    Task t = tasks.get(taskid);
-    if (!savedTaskOutput) {
-      t.saveTaskOutput();
-      savedTaskOutput = true;
-    } else {
-      try {
-        t.discardTaskOutput();
-      } catch (IOException ioe) {
-        LOG.info("Failed to discard 'already-saved' output of task: " + 
-                t.getTaskId() + " with: " + 
-                StringUtils.stringifyException(ioe));
-      }
-    }
-
+  public void completed(String taskid) {
     //
     // Record that this taskid is complete
     //
     completedTask(taskid, TaskStatus.State.SUCCEEDED);
-    
         
     //
     // Now that the TIP is complete, the other speculative 
@@ -545,7 +538,6 @@ class TaskInProgress {
     this.completes++;
     recomputeProgress();
     
-    LOG.info("Task '" + taskid + "' has completed succesfully");
   }
 
   /**
@@ -588,7 +580,8 @@ class TaskInProgress {
    */
   boolean killTask(String taskId, boolean shouldFail) {
     TaskStatus st = taskStatuses.get(taskId);
-    if(st != null && st.getRunState() == TaskStatus.State.RUNNING
+    if(st != null && (st.getRunState() == TaskStatus.State.RUNNING
+        || st.getRunState() == TaskStatus.State.COMMIT_PENDING)
         && tasksToKill.put(taskId, shouldFail) == null ) {
       String logStr = "Request received to " + (shouldFail ? "fail" : "kill") 
                       + " task '" + taskId + "' by user";
@@ -599,32 +592,6 @@ class TaskInProgress {
     return false;
   }
 
-  /** Notification that a task with the given id has been killed */
-  void taskKilled(String taskId, String trackerName, JobStatus jobStatus) {
-    Boolean shouldFail = tasksToKill.remove(taskId);
-    if(shouldFail != null && !shouldFail) {
-      LOG.info("Task '" + taskId + "' has been killed");
-      this.activeTasks.remove(taskId);
-      taskStatuses.get(taskId).setRunState(TaskStatus.State.KILLED );
-      addDiagnosticInfo(taskId, "Task has been killed" );
-      // Discard task output
-      Task t = tasks.get(taskId);
-      try {
-        t.discardTaskOutput();
-      } catch (IOException ioe) {
-        LOG.info("Failed to discard output of task '" + taskId + "' with " +
-            StringUtils.stringifyException(ioe));
-      }
-      numKilledTasks++;
-      
-    }
-    else {
-      //set the task status as failed. 
-      taskStatuses.get(taskId).setRunState(TaskStatus.State.FAILED);
-      incompleteSubTask(taskId, trackerName, jobStatus);
-    }
-  }
-
   /**
    * This method is called whenever there's a status change
    * for one of the TIP's sub-tasks.  It recomputes the overall 
@@ -650,6 +617,12 @@ class TaskInProgress {
           bestState = status.getStateString();
           bestCounters = status.getCounters();
           break;
+        } else if (status.getRunState() == TaskStatus.State.COMMIT_PENDING) {
+          //for COMMIT_PENDING, we take the last state that we recorded
+          //when the task was RUNNING
+          bestProgress = this.progress;
+          bestState = this.state;
+          bestCounters = this.counters;
         } else if (status.getRunState() == TaskStatus.State.RUNNING) {
           if (status.getProgress() >= bestProgress) {
             bestProgress = status.getProgress();
@@ -692,7 +665,7 @@ class TaskInProgress {
         runSpeculative &&
         (averageProgress - progress >= SPECULATIVE_GAP) &&
         (System.currentTimeMillis() - startTime >= SPECULATIVE_LAG) 
-        && completes == 0) {
+        && completes == 0 && !isOnlyCommitPending()) {
       return true;
     }
     return false;

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

@@ -40,7 +40,8 @@ abstract class TaskStatus implements Writable, Cloneable {
   public static enum Phase{STARTING, MAP, SHUFFLE, SORT, REDUCE}
 
   // what state is the task in?
-  public static enum State {RUNNING, SUCCEEDED, FAILED, UNASSIGNED, KILLED}
+  public static enum State {RUNNING, SUCCEEDED, FAILED, UNASSIGNED, KILLED, 
+                            COMMIT_PENDING}
     
   private String taskid;
   private float progress;

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

@@ -1368,7 +1368,7 @@ public class TaskTracker
      * The task is reporting that it's done running
      */
     public synchronized void reportDone() {
-      this.taskStatus.setRunState(TaskStatus.State.SUCCEEDED);
+      this.taskStatus.setRunState(TaskStatus.State.COMMIT_PENDING);
       this.taskStatus.setProgress(1.0f);
       this.taskStatus.setFinishTime(System.currentTimeMillis());
       this.done = true;
@@ -1400,7 +1400,7 @@ public class TaskTracker
       boolean needCleanup = false;
       synchronized (this) {
         if (done) {
-          taskStatus.setRunState(TaskStatus.State.SUCCEEDED);
+          taskStatus.setRunState(TaskStatus.State.COMMIT_PENDING);
         } else {
           if (!wasKilled) {
             failures += 1;
@@ -1477,7 +1477,10 @@ public class TaskTracker
      */
     private synchronized void mapOutputLost(String failure
                                            ) throws IOException {
-      if (taskStatus.getRunState() == TaskStatus.State.SUCCEEDED) {
+      //The check for COMMIT_PENDING should actually be a check for SUCCESS
+      //however for that, we have to introduce another Action type from the
+      //JT to the TT (SuccessTaskAction in the lines of KillTaskAction).
+      if (taskStatus.getRunState() == TaskStatus.State.COMMIT_PENDING) {
         // change status to failure
         LOG.info("Reporting output lost:"+task.getTaskId());
         taskStatus.setRunState(TaskStatus.State.FAILED);

src/webapps/job/taskdetails.jsp

@@ -155,7 +155,7 @@
         }
         out.print("</td><td>" + "<a href=\"/taskstats.jsp?jobid=" + jobid
           + "&tipid=" + tipid + "&taskid=" + status.getTaskId() + "\">"
-          + status.getCounters().size() + "</a></td>");
+          + ((status.getCounters() != null) ? status.getCounters().size() : 0) + "</a></td>");
         out.print("<td>");
         if (privateActions
           && status.getRunState() == TaskStatus.State.RUNNING) {