Meerge -r 704535:704536 from trunk onto 0.19 branch. Fixes HADOOP-4361.

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

CHANGES.txt

@@ -844,6 +844,10 @@ Release 0.19.0 - Unreleased
     HADOOP-4287. Fixes an issue to do with maintaining counts of running/pending
     maps/reduces. (Sreekanth Ramakrishnan via ddas)
 
+    HADOOP-4361. Makes sure that jobs killed from command line are killed
+    fast (i.e., there is a slot to run the cleanup task soon).
+    (Amareshwari Sriramadasu via ddas)
+
 Release 0.18.2 - Unreleased
 
   BUG FIXES

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

@@ -858,7 +858,7 @@ class JobInProgress {
                                             int clusterSize, 
                                             int numUniqueHosts
                                            ) throws IOException {
-    if (!tasksInited.get()) {
+    if (status.getRunState() != JobStatus.RUNNING) {
       LOG.info("Cannot create task split for " + profile.getJobID());
       return null;
     }
@@ -1016,7 +1016,7 @@ class JobInProgress {
                                                int clusterSize,
                                                int numUniqueHosts
                                               ) throws IOException {
-    if (!tasksInited.get()) {
+    if (status.getRunState() != JobStatus.RUNNING) {
       LOG.info("Cannot create task split for " + profile.getJobID());
       return null;
     }
@@ -1918,6 +1918,9 @@ class JobInProgress {
       //
       // kill all TIPs.
       //
+      for (int i = 0; i < setup.length; i++) {
+        setup[i].kill();
+      }
       for (int i = 0; i < maps.length; i++) {
         maps[i].kill();
       }

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

@@ -2054,7 +2054,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
           // It may be successfully completed, or may be killed in
           // mid-execution.
           //
-          if (tip.getJob().getStatus().getRunState() == JobStatus.RUNNING) {
+          if (tip.getJob().getStatus().getRunState() == JobStatus.RUNNING ||
+              tip.getJob().getStatus().getRunState() == JobStatus.PREP) {
             killList.add(new KillTaskAction(killTaskId));
             LOG.debug(taskTracker + " -> KillTaskAction: " + killTaskId);
           } else {
@@ -2619,7 +2620,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
             (tip.isMapTask() && !tip.isSetupTask() && 
              job.desiredReduces() != 0)) {
           // if the job is done, we don't want to change anything
-          if (job.getStatus().getRunState() == JobStatus.RUNNING) {
+          if (job.getStatus().getRunState() == JobStatus.RUNNING ||
+              job.getStatus().getRunState() == JobStatus.PREP) {
             job.failedTask(tip, taskId, ("Lost task tracker: " + trackerName), 
                            (tip.isMapTask() ? 
                                TaskStatus.Phase.MAP : 

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

@@ -386,7 +386,9 @@ class TaskInProgress {
     TaskStatus ts = taskStatuses.get(taskid);
     if ((ts != null) &&
         (!tasksReportedClosed.contains(taskid)) &&
-        (job.getStatus().getRunState() != JobStatus.RUNNING)) {
+        ((this.failed) ||
+        ((job.getStatus().getRunState() != JobStatus.RUNNING &&
+         (job.getStatus().getRunState() != JobStatus.PREP))))) {
       tasksReportedClosed.add(taskid);
       close = true;
     } else if (isComplete() &&