commit b3b11c4936a501eda5991fbfa4498fc1f336cf9d

Author: Lee Tucker <ltucker@yahoo-inc.com>
Date:   Thu Jul 30 17:40:17 2009 -0700

    Applying patch 2709938.5739.patch


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-patches@1076933 13f79535-47bb-0310-9956-ffa450edef68

src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/ClusterWithCapacityScheduler.java

@@ -197,6 +197,13 @@ public class ClusterWithCapacityScheduler extends TestCase {
     }
   }
 
+  /**
+   * @return the mrCluster
+   */
+  public MiniMRCluster getMrCluster() {
+    return mrCluster;
+  }
+
   static class MyClassLoader extends ClassLoader {
     @Override
     public URL getResource(String name) {

src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestJobTrackerRestartWithCS.java

@@ -0,0 +1,82 @@
+/**
+ * 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 java.util.Properties;
+import org.apache.hadoop.mapred.ControlledMapReduceJob.ControlledMapReduceJobRunner;
+
+
+public class TestJobTrackerRestartWithCS extends ClusterWithCapacityScheduler {
+
+  /**
+   * Test single queue.
+   *
+   * <p>
+   *
+   * Submit a job with more M/R tasks than total capacity. Full queue capacity
+   * should be utilized and remaining M/R tasks should wait for slots to be
+   * available.
+   *
+   * @throws Exception
+   */
+  public void testJobTrackerRestartWithCS()
+          throws Exception {
+    try {
+      Properties schedulerProps = new Properties();
+      schedulerProps.put(
+              "mapred.capacity-scheduler.queue.default.guaranteed-capacity", "100");
+      Properties clusterProps = new Properties();
+      clusterProps.put("mapred.tasktracker.map.tasks.maximum", String.valueOf(2));
+      clusterProps.put("mapred.tasktracker.reduce.tasks.maximum", String.valueOf(0));
+
+      // cluster capacity 2 maps, 0 reduces
+      startCluster(1, clusterProps, schedulerProps);
+
+      ControlledMapReduceJobRunner jobRunner =
+              ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+              getJobConf(), 4, 0);
+      jobRunner.start();
+      ControlledMapReduceJob controlledJob = jobRunner.getJob();
+      JobID myJobID = jobRunner.getJobID();
+      JobInProgress myJob = getJobTracker().getJob(myJobID);
+      ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 2);
+
+      LOG.info("Trying to finish 2 maps");
+      controlledJob.finishNTasks(true, 2);
+      ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 2);
+      assertTrue("Number of maps finished", myJob.finishedMaps() == 2);
+
+      JobClient jobClient = new JobClient(getMrCluster().createJobConf());
+      getMrCluster().stopJobTracker();
+
+      getMrCluster().getJobTrackerConf().setBoolean("mapred.jobtracker.restart.recover",
+              true);
+      getMrCluster().startJobTracker();
+
+      UtilsForTests.waitForJobTracker(jobClient);
+      ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 1);
+
+      controlledJob.finishNTasks(true, 2);
+      ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 2);
+    } catch (Exception e) {
+      e.printStackTrace();
+    } finally {
+      tearDown();
+    }
+  }
+}

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

@@ -739,7 +739,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         
         TaskID id = TaskID.forName(taskId);
         TaskInProgress tip = getTip(id);
-        
+
         updateTip(tip, task);
       }
 
@@ -752,7 +752,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         
         // Check if the transaction for this attempt can be committed
         String taskStatus = attempt.get(Keys.TASK_STATUS);
-        
+        TaskAttemptID taskID = TaskAttemptID.forName(taskAttemptId);
+        JobInProgress jip = getJob(taskID.getJobID());
+        JobStatus prevStatus = (JobStatus)jip.getStatus().clone();
+
         if (taskStatus.length() > 0) {
           // This means this is an update event
           if (taskStatus.equals(Values.SUCCESS.name())) {
@@ -766,6 +769,16 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         } else {
           createTaskAttempt(jip, id, attempt);
         }
+        
+        JobStatus newStatus = (JobStatus)jip.getStatus().clone();
+        if (prevStatus.getRunState() != newStatus.getRunState()) {
+          if(LOG.isDebugEnabled())
+            LOG.debug("Status changed hence informing prevStatus" +  prevStatus + " currentStatus "+ newStatus);
+          JobStatusChangeEvent event =
+            new JobStatusChangeEvent(jip, EventType.RUN_STATE_CHANGED,
+                                     prevStatus, newStatus);
+          updateJobInProgressListeners(event);
+        }
       }
 
       public void handle(JobHistory.RecordTypes recType, Map<Keys,