svn commit: r1311966 - in /hadoop/common/branches/branch-1: CHANGES.txt src/mapred/org/apache/hadoop/mapred/JobInProgress.java

[bobby]

Author: bobby
Date: Tue Apr 10 20:14:23 2012
New Revision: 1311966

URL: http://svn.apache.org/viewvc?rev=1311966&view=rev
Log:
MAPREDUCE-1238. mapred metrics shows negative count of waiting maps and reduces 
(tgraves via bobby)

Modified:
    hadoop/common/branches/branch-1/CHANGES.txt
    
hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobInProgress.java

Modified: hadoop/common/branches/branch-1/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/CHANGES.txt?rev=1311966&r1=1311965&r2=1311966&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/CHANGES.txt (original)
+++ hadoop/common/branches/branch-1/CHANGES.txt Tue Apr 10 20:14:23 2012
@@ -205,6 +205,9 @@ Release 1.1.0 - unreleased
     HDFS-3176. Use MD5MD5CRC32FileChecksum.readFields() in JsonUtil .  (Kihwal
     Lee via szetszwo)
 
+    MAPREDUCE-1238. mapred metrics shows negative count of waiting maps and
+    reduces (tgraves via bobby)
+
 Release 1.0.3 - unreleased
 
   NEW FEATURES

Modified: 
hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
URL: 
http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobInProgress.java?rev=1311966&r1=1311965&r2=1311966&view=diff
==============================================================================
--- 
hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
 (original)
+++ 
hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
 Tue Apr 10 20:14:23 2012
@@ -732,7 +732,7 @@ public class JobInProgress {
     if (numMapTasks > 0) { 
       nonRunningMapCache = createCache(splits, maxLevel);
     }
-        
+
     // set the launch time
     this.launchTime = jobtracker.getClock().getTime();
 
@@ -789,12 +789,15 @@ public class JobInProgress {
     
     synchronized(jobInitKillStatus){
       jobInitKillStatus.initDone = true;
+
+      // set this before the throw to make sure cleanup works properly
+      tasksInited = true;
+
       if(jobInitKillStatus.killed) {
         throw new KillInterruptedException("Job " + jobId + " killed in init");
       }
     }
     
-    tasksInited = true;
     JobHistory.JobInfo.logInited(profile.getJobID(), this.launchTime, 
                                  numMapTasks, numReduceTasks);
     
@@ -3225,11 +3228,16 @@ public class JobInProgress {
       // Cancel task tracker reservation
       cancelReservedSlots();
 
+      //  Waiting metrics are incremented in JobInProgress.initTasks()
+      //  If a job gets an exception before that, we do not want to
+      //  incorrectly decrement.
+      if (tasksInited) {
+        jobtracker.getInstrumentation().decWaitingMaps(getJobID(), 
pendingMaps());
+        jobtracker.getInstrumentation().decWaitingReduces(getJobID(), 
pendingReduces());
+        this.queueMetrics.decWaitingMaps(getJobID(), pendingMaps());
+        this.queueMetrics.decWaitingReduces(getJobID(), pendingReduces());
+      }
       // Let the JobTracker know that a job is complete
-      jobtracker.getInstrumentation().decWaitingMaps(getJobID(), 
pendingMaps());
-      jobtracker.getInstrumentation().decWaitingReduces(getJobID(), 
pendingReduces());
-      this.queueMetrics.decWaitingMaps(getJobID(), pendingMaps());
-      this.queueMetrics.decWaitingReduces(getJobID(), pendingReduces());
       jobtracker.storeCompletedJob(this);
       jobtracker.finalizeJob(this);