Merge -r 732230:732231 from trunk to branch 0.20 to fix HADOOP-4830.

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

CHANGES.txt

@@ -289,6 +289,9 @@ Release 0.20.0 - Unreleased
     HADOOP-4909. Fix Javadoc and make some of the API more consistent in their
     use of the JobContext instead of Configuration. (omalley)
 
+    HADOOP-4830. Add end-to-end test cases for testing queue capacities.
+    (Vinod Kumar Vavilapalli via yhemanth)
+
   OPTIMIZATIONS
 
     HADOOP-3293. Fixes FileInputFormat to do provide locations for splits

src/contrib/capacity-scheduler/ivy.xml

@@ -36,5 +36,21 @@
       name="log4j"
       rev="${log4j.version}"
       conf="common->master"/>
-    </dependencies>
+    <dependency org="org.mortbay.jetty"
+      name="jetty-util"
+      rev="${jetty-util.version}"
+      conf="common->master"/>
+    <dependency org="org.mortbay.jetty"
+      name="jetty"
+      rev="${jetty.version}"
+      conf="common->master"/>
+    <dependency org="org.mortbay.jetty"
+      name="servlet-api-2.5"
+      rev="${servlet-api-2.5.version}"
+      conf="common->master"/> 
+    <dependency org="commons-httpclient"
+      name="commons-httpclient"
+      rev="${commons-httpclient.version}"
+      conf="common->master"/> 
+  </dependencies>
 </ivy-module>

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

@@ -286,7 +286,7 @@ class CapacitySchedulerConf {
     initializeDefaults();
   }
   
-  private static final String toFullPropertyName(String queue, 
+  static final String toFullPropertyName(String queue, 
                                                   String property) {
       return QUEUE_CONF_PROPERTY_NAME_PREFIX + queue + "." + property;
   }

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

@@ -0,0 +1,240 @@
+/**
+ * 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.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Enumeration;
+import java.util.Properties;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+
+/**
+ * A test-cluster based on {@link MiniMRCluster} that is started with
+ * CapacityTaskScheduler. It provides knobs to configure both the cluster as
+ * well as the scheduler. Any test that intends to test capacity-scheduler
+ * should extend this.
+ * 
+ */
+public class ClusterWithCapacityScheduler extends TestCase {
+
+  static final Log LOG = LogFactory.getLog(ClusterWithCapacityScheduler.class);
+  private MiniMRCluster mrCluster;
+  private MiniDFSCluster dfsCluster;
+
+  private JobConf jobConf;
+
+  static final String MY_SCHEDULER_CONF_PATH_PROPERTY = "my.resource.path";
+
+  protected void startCluster()
+      throws IOException {
+    startCluster(null, null);
+  }
+
+  /**
+   * Start the cluster with two TaskTrackers and two DataNodes and configure the
+   * cluster with clusterProperties and the scheduler with schedulerProperties.
+   * Uses default configuration whenever user provided properties are missing
+   * (null/empty)
+   * 
+   * @param clusterProperties
+   * @param schedulerProperties
+   * @throws IOException
+   */
+  protected void startCluster(Properties clusterProperties,
+      Properties schedulerProperties)
+      throws IOException {
+    startCluster(2, 2, clusterProperties, schedulerProperties);
+  }
+
+  /**
+   * Start the cluster with numTaskTrackers TaskTrackers and numDataNodes
+   * DataNodes and configure the cluster with clusterProperties and the
+   * scheduler with schedulerProperties. Uses default configuration whenever
+   * user provided properties are missing (null/empty)
+   * 
+   * @param numTaskTrackers
+   * @param numDataNodes
+   * @param clusterProperties
+   * @param schedulerProperties
+   * @throws IOException
+   */
+  protected void startCluster(int numTaskTrackers, int numDataNodes,
+      Properties clusterProperties, Properties schedulerProperties)
+      throws IOException {
+    Thread.currentThread().setContextClassLoader(
+        new ClusterWithCapacityScheduler.MyClassLoader());
+    JobConf clusterConf = new JobConf();
+    if (clusterProperties != null) {
+      for (Enumeration<?> e = clusterProperties.propertyNames(); e
+          .hasMoreElements();) {
+        String key = (String) e.nextElement();
+        clusterConf.set(key, (String) clusterProperties.get(key));
+      }
+    }
+    dfsCluster = new MiniDFSCluster(clusterConf, numDataNodes, true, null);
+
+    if (schedulerProperties != null) {
+      setUpSchedulerConfigFile(schedulerProperties);
+    }
+
+    clusterConf.set("mapred.jobtracker.taskScheduler",
+        CapacityTaskScheduler.class.getName());
+    mrCluster =
+        new MiniMRCluster(numTaskTrackers, dfsCluster.getFileSystem().getUri()
+            .toString(), 1, null, null, clusterConf);
+
+    this.jobConf = mrCluster.createJobConf(clusterConf);
+  }
+
+  private void setUpSchedulerConfigFile(Properties schedulerConfProps)
+      throws IOException {
+    Configuration config = new Configuration(false);
+
+    LocalFileSystem fs = new LocalFileSystem();
+    fs.setConf(config);
+    // The above call doesn't set the configuration for the underlying
+    // RawFileSystem. Explicitly doing it.
+    fs.getRawFileSystem().setConf(config);
+
+    String myResourcePath = System.getProperty("test.build.data");
+    Path schedulerConfigFilePath =
+        new Path(myResourcePath, CapacitySchedulerConf.SCHEDULER_CONF_FILE);
+    OutputStream out = fs.create(schedulerConfigFilePath);
+
+    for (Enumeration<?> e = schedulerConfProps.propertyNames(); e
+        .hasMoreElements();) {
+      String key = (String) e.nextElement();
+      LOG.debug("Adding " + key + schedulerConfProps.getProperty(key));
+      config.set(key, schedulerConfProps.getProperty(key));
+    }
+
+    config.writeXml(out);
+    out.close();
+
+    LOG.info("setting resource path where capacity-scheduler's config file "
+        + "is placed to " + myResourcePath);
+    System.setProperty(MY_SCHEDULER_CONF_PATH_PROPERTY, myResourcePath);
+  }
+
+  private void cleanUpSchedulerConfigFile() throws IOException {
+    Configuration config = new Configuration(false);
+
+    LocalFileSystem fs = new LocalFileSystem();
+    fs.setConf(config);
+    fs.getRawFileSystem().setConf(config);
+
+    String myResourcePath = System.getProperty("test.build.data");
+    Path schedulerConfigFilePath =
+        new Path(myResourcePath, CapacitySchedulerConf.SCHEDULER_CONF_FILE);
+    fs.delete(schedulerConfigFilePath, false);
+  }
+
+  protected JobConf getJobConf() {
+    return this.jobConf;
+  }
+
+  protected JobTracker getJobTracker() {
+    return this.mrCluster.getJobTrackerRunner().getJobTracker();
+  }
+
+  @Override
+  protected void tearDown()
+      throws Exception {
+    cleanUpSchedulerConfigFile();
+    
+    if (mrCluster != null) {
+      mrCluster.shutdown();
+    }
+    if (dfsCluster != null) {
+      dfsCluster.shutdown();
+    }
+  }
+
+  /**
+   * Wait till all the slots in the cluster are occupied with respect to the
+   * tasks of type specified isMap.
+   * 
+   * <p>
+   * 
+   * <b>Also, it is assumed that the tasks won't finish any time soon, like in
+   * the case of tasks of {@link ControlledMapReduceJob}</b>.
+   * 
+   * @param isMap
+   */
+  protected void waitTillAllSlotsAreOccupied(boolean isMap)
+      throws InterruptedException {
+    JobTracker jt = this.mrCluster.getJobTrackerRunner().getJobTracker();
+    ClusterStatus clusterStatus = jt.getClusterStatus();
+    int currentTasks =
+        (isMap ? clusterStatus.getMapTasks() : clusterStatus.getReduceTasks());
+    int maxTasks =
+        (isMap ? clusterStatus.getMaxMapTasks() : clusterStatus
+            .getMaxReduceTasks());
+    while (currentTasks != maxTasks) {
+      Thread.sleep(1000);
+      clusterStatus = jt.getClusterStatus();
+      currentTasks =
+          (isMap ? clusterStatus.getMapTasks() : clusterStatus
+              .getReduceTasks());
+      maxTasks =
+          (isMap ? clusterStatus.getMaxMapTasks() : clusterStatus
+              .getMaxReduceTasks());
+      LOG.info("Waiting till cluster reaches steady state. currentTasks : "
+          + currentTasks + " total cluster capacity : " + maxTasks);
+    }
+  }
+
+  static class MyClassLoader extends ClassLoader {
+    @Override
+    public URL getResource(String name) {
+      if (!name.equals(CapacitySchedulerConf.SCHEDULER_CONF_FILE)) {
+        return super.getResource(name);
+      }
+      return findResource(name);
+    }
+
+    @Override
+    protected URL findResource(String name) {
+      try {
+        String resourcePath =
+            System
+                .getProperty(ClusterWithCapacityScheduler.MY_SCHEDULER_CONF_PATH_PROPERTY);
+        // Check the resourcePath directory
+        File file = new File(resourcePath, name);
+        if (file.exists()) {
+          return new URL("file://" + file.getAbsolutePath());
+        }
+      } catch (MalformedURLException mue) {
+        LOG.warn("exception : " + mue);
+      }
+      return super.findResource(name);
+    }
+  }
+}

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

@@ -0,0 +1,440 @@
+/**
+ * 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;
+
+/**
+ * End to end tests based on MiniMRCluster to verify that queue capacities are
+ * honored. Automates the tests related to queue capacities: submits jobs to
+ * different queues simultaneously and ensures that capacities are honored
+ */
+public class TestQueueCapacities 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 testSingleQueue()
+      throws Exception {
+
+    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(3));
+    clusterProps.put("mapred.tasktracker.reduce.tasks.maximum", String
+        .valueOf(3));
+    // cluster capacity 12 maps, 12 reduces
+    startCluster(4, 2, clusterProps, schedulerProps);
+
+    ControlledMapReduceJobRunner jobRunner =
+        ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+            getJobConf(), 16, 16);
+    jobRunner.start();
+    ControlledMapReduceJob controlledJob = jobRunner.getJob();
+    JobID myJobID = jobRunner.getJobID();
+    JobInProgress myJob = getJobTracker().getJob(myJobID);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 12);
+
+    // Wait till the cluster reaches steady state. This confirms that the rest
+    // of the tasks are not running and waiting for slots
+    // to be freed.
+    waitTillAllSlotsAreOccupied(true);
+
+    LOG.info("Trying to finish 2 maps");
+    controlledJob.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 2);
+    assertTrue("Number of maps finished", myJob.finishedMaps() == 2);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 12);
+    waitTillAllSlotsAreOccupied(true);
+
+    LOG.info("Trying to finish 2 more maps");
+    controlledJob.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 4);
+    assertTrue("Number of maps finished", myJob.finishedMaps() == 4);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 12);
+    waitTillAllSlotsAreOccupied(true);
+
+    LOG.info("Trying to finish the last 12 maps");
+    controlledJob.finishNTasks(true, 12);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 16);
+    assertTrue("Number of maps finished", myJob.finishedMaps() == 16);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 0);
+    ControlledMapReduceJob.haveAllTasksFinished(myJob, true);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, false, 12);
+    waitTillAllSlotsAreOccupied(false);
+
+    LOG.info("Trying to finish 4 reduces");
+    controlledJob.finishNTasks(false, 4);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, false, 4);
+    assertTrue("Number of reduces finished", myJob.finishedReduces() == 4);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, false, 12);
+    waitTillAllSlotsAreOccupied(false);
+
+    LOG.info("Trying to finish the last 12 reduces");
+    controlledJob.finishNTasks(false, 12);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, false, 16);
+    assertTrue("Number of reduces finished", myJob.finishedReduces() == 16);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, false, 0);
+    ControlledMapReduceJob.haveAllTasksFinished(myJob, false);
+
+    jobRunner.join();
+  }
+
+  /**
+   * Test single queue with multiple jobs.
+   * 
+   * @throws Exception
+   */
+  public void testSingleQueueMultipleJobs()
+      throws Exception {
+
+    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(3));
+    clusterProps.put("mapred.tasktracker.reduce.tasks.maximum", String
+        .valueOf(0));
+    // cluster capacity 12 maps, 0 reduces
+    startCluster(4, 2, clusterProps, schedulerProps);
+
+    singleQMultipleJobs1();
+    singleQMultipleJobs2();
+  }
+
+  /**
+   * Test multiple queues.
+   * 
+   * These tests use 4 queues default, Q2, Q3 and Q4 with guaranteed capacities
+   * 10, 20, 30, 40 respectively), user limit 100%, priority not respected, one
+   * user per queue. Reclaim time 5 minutes.
+   * 
+   * @throws Exception
+   */
+  public void testMultipleQueues()
+      throws Exception {
+    Properties schedulerProps = new Properties();
+    String[] queues = new String[] { "default", "Q2", "Q3", "Q4" };
+    int GC = 0;
+    for (String q : queues) {
+      GC += 10;
+      schedulerProps.put(CapacitySchedulerConf.toFullPropertyName(q,
+          "guaranteed-capacity"), String.valueOf(GC)); // TODO: use strings
+      schedulerProps.put(CapacitySchedulerConf.toFullPropertyName(q,
+          "minimum-user-limit-percent"), String.valueOf(100));
+      schedulerProps.put(CapacitySchedulerConf.toFullPropertyName(q,
+          "reclaim-time-limit"), String.valueOf(300));
+    }
+
+    Properties clusterProps = new Properties();
+    clusterProps
+        .put("mapred.tasktracker.map.tasks.maximum", String.valueOf(2));
+    clusterProps.put("mapred.tasktracker.reduce.tasks.maximum", String
+        .valueOf(2));
+    clusterProps.put("mapred.queue.names", queues[0] + "," + queues[1] + ","
+        + queues[2] + "," + queues[3]);
+
+    // cluster capacity 10 maps, 10 reduces and 4 queues with capacities 1, 2,
+    // 3, 4 respectively.
+    startCluster(5, 2, clusterProps, schedulerProps);
+
+    multipleQsWithOneQBeyondCapacity(queues);
+    multipleQueuesWithinCapacities(queues);
+  }
+
+  /**
+   * Submit a job with more M/R tasks than total queue capacity and then submit
+   * another job. First job utilizes all the slots. When the second job is
+   * submitted, the tasks of the second job wait for slots to be available. As
+   * the tasks of the first jobs finish and there are no more tasks pending, the
+   * tasks of the second job start running on the freed up slots.
+   * 
+   * @throws Exception
+   */
+  private void singleQMultipleJobs1()
+      throws Exception {
+
+    ControlledMapReduceJobRunner jobRunner1 =
+        ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+            getJobConf(), 16, 0);
+    ControlledMapReduceJobRunner jobRunner2 =
+        ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+            getJobConf(), 12, 0);
+    jobRunner1.start();
+    ControlledMapReduceJob controlledJob1 = jobRunner1.getJob();
+    JobID jobID1 = jobRunner1.getJobID();
+    JobInProgress jip1 = getJobTracker().getJob(jobID1);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip1, true, 12);
+
+    // Confirm that the rest of the tasks are not running and waiting for slots
+    // to be freed.
+    waitTillAllSlotsAreOccupied(true);
+
+    // Now start the second job.
+    jobRunner2.start();
+    JobID jobID2 = jobRunner2.getJobID();
+    ControlledMapReduceJob controlledJob2 = jobRunner2.getJob();
+    JobInProgress jip2 = getJobTracker().getJob(jobID2);
+
+    LOG.info("Trying to finish 2 map");
+    controlledJob1.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip1, true, 2);
+    assertTrue("Number of maps finished", jip1.finishedMaps() == 2);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip1, true, 12);
+    waitTillAllSlotsAreOccupied(true);
+
+    LOG.info("Trying to finish 2 more maps");
+    controlledJob1.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip1, true, 4);
+    assertTrue("Number of maps finished", jip1.finishedMaps() == 4);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip1, true, 12);
+    waitTillAllSlotsAreOccupied(true);
+
+    // All tasks of Job1 started running/finished. Now job2 should start
+    LOG.info("Trying to finish 2 more maps");
+    controlledJob1.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip1, true, 6);
+    assertTrue("Number of maps finished", jip1.finishedMaps() == 6);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip1, true, 10);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip2, true, 2);
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(jip1, true, 10);
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 2);
+
+    LOG.info("Trying to finish 10 more maps and hence job1");
+    controlledJob1.finishNTasks(true, 10);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip1, true, 16);
+    assertTrue("Number of maps finished", jip1.finishedMaps() == 16);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip2, true, 12);
+    controlledJob1.finishJob();
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(jip1, true, 0);
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 12);
+
+    // Finish job2 also
+    controlledJob2.finishJob();
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip2, true, 12);
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 0);
+
+    jobRunner1.join();
+    jobRunner2.join();
+  }
+
+  /**
+   * Submit a job with less M/R tasks than total capacity and another job with
+   * more M/R tasks than the remaining capacity. First job should utilize the
+   * required slots and other job should utilize the available slots and its
+   * remaining tasks wait for slots to become free.
+   * 
+   * @throws Exception
+   */
+  private void singleQMultipleJobs2()
+      throws Exception {
+
+    ControlledMapReduceJobRunner jobRunner1 =
+        ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+            getJobConf(), 8, 0);
+    ControlledMapReduceJobRunner jobRunner2 =
+        ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+            getJobConf(), 12, 0);
+    jobRunner1.start();
+    ControlledMapReduceJob controlledJob1 = jobRunner1.getJob();
+    JobID jobID1 = jobRunner1.getJobID();
+    JobInProgress jip1 = getJobTracker().getJob(jobID1);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip1, true, 8);
+    ControlledMapReduceJob.assertNumTasksRunning(jip1, true, 8);
+
+    // Now start the second job.
+    jobRunner2.start();
+    JobID jobID2 = jobRunner2.getJobID();
+    ControlledMapReduceJob controlledJob2 = jobRunner2.getJob();
+    JobInProgress jip2 = getJobTracker().getJob(jobID2);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip2, true, 4);
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(jip1, true, 8);
+    // The rest of the tasks of job2 should wait.
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 4);
+
+    LOG.info("Trying to finish 2 maps of job1");
+    controlledJob1.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip1, true, 2);
+    assertTrue("Number of maps finished", jip1.finishedMaps() == 2);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip1, true, 6);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip2, true, 6);
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(jip1, true, 6);
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 6);
+
+    LOG.info("Trying to finish 6 more maps of job1");
+    controlledJob1.finishNTasks(true, 6);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip1, true, 8);
+    assertTrue("Number of maps finished", jip1.finishedMaps() == 8);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip2, true, 12);
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(jip1, true, 0);
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 12);
+
+    // Finish job2 also
+    controlledJob2.finishJob();
+
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip2, true, 12);
+    ControlledMapReduceJob.assertNumTasksRunning(jip2, true, 0);
+
+    jobRunner1.join();
+    jobRunner2.join();
+  }
+
+  /**
+   * Test to verify running of tasks in a queue going over its capacity. In
+   * queue default, user U1 starts a job J1, having more M/R tasks than the
+   * total slots. M/R tasks of job J1 should start running on all the nodes (100
+   * % utilization).
+   * 
+   * @throws Exception
+   */
+  private void multipleQsWithOneQBeyondCapacity(String[] queues)
+      throws Exception {
+
+    JobConf conf = getJobConf();
+    conf.setQueueName(queues[0]);
+    conf.setUser("U1");
+    ControlledMapReduceJobRunner jobRunner =
+        ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(conf, 15,
+            0);
+    jobRunner.start();
+    ControlledMapReduceJob controlledJob = jobRunner.getJob();
+    JobID myJobID = jobRunner.getJobID();
+    JobInProgress myJob = getJobTracker().getJob(myJobID);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 10);
+
+    // Confirm that the rest of the tasks are not running and waiting for slots
+    // to be freed.
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(myJob, true, 10);
+
+    LOG.info("Trying to finish 3 maps");
+    controlledJob.finishNTasks(true, 3);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 3);
+    assertTrue("Number of maps finished", myJob.finishedMaps() == 3);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 10);
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(myJob, true, 10);
+
+    LOG.info("Trying to finish 2 more maps");
+    controlledJob.finishNTasks(true, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 5);
+    assertTrue("Number of maps finished", myJob.finishedMaps() == 5);
+    ControlledMapReduceJob.waitTillNTasksStartRunning(myJob, true, 10);
+    waitTillAllSlotsAreOccupied(true);
+    ControlledMapReduceJob.assertNumTasksRunning(myJob, true, 10);
+
+    // Finish job
+    controlledJob.finishJob();
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(myJob, true, 15);
+    ControlledMapReduceJob.assertNumTasksRunning(myJob, true, 0);
+    jobRunner.join();
+  }
+
+  /**
+   * Test to verify queue capacities across multiple queues. In this test, jobs
+   * are submitted to different queues - all below the queue's capacity and
+   * verifies that all the jobs are running. This will test code paths related
+   * to job initialization, considering multiple queues for scheduling jobs etc.
+   * 
+   * <p>
+   * 
+   * One user per queue. Four jobs are submitted to the four queues such that
+   * they exactly fill up the queues. No queue should be beyond capacity. All
+   * jobs should be running.
+   * 
+   * @throws Exception
+   */
+  private void multipleQueuesWithinCapacities(String[] queues)
+      throws Exception {
+    String[] users = new String[] { "U1", "U2", "U3", "U4" };
+    ControlledMapReduceJobRunner[] jobRunners =
+        new ControlledMapReduceJobRunner[4];
+    ControlledMapReduceJob[] controlledJobs = new ControlledMapReduceJob[4];
+    JobInProgress[] jips = new JobInProgress[4];
+
+    // Initialize all the jobs
+    // Start all the jobs in parallel
+    JobConf conf = getJobConf();
+    int numTasks = 1;
+    for (int i = 0; i < 4; i++) {
+      conf.setQueueName(queues[i]);
+      conf.setUser(users[i]);
+      jobRunners[i] =
+          ControlledMapReduceJobRunner.getControlledMapReduceJobRunner(
+              getJobConf(), numTasks, numTasks);
+      jobRunners[i].start();
+      controlledJobs[i] = jobRunners[i].getJob();
+      JobID jobID = jobRunners[i].getJobID();
+      jips[i] = getJobTracker().getJob(jobID);
+      // Wait till all the jobs start running all of their tasks
+      ControlledMapReduceJob.waitTillNTasksStartRunning(jips[i], true,
+          numTasks);
+      ControlledMapReduceJob.waitTillNTasksStartRunning(jips[i], false,
+          numTasks);
+      numTasks += 1;
+    }
+
+    // Ensure steady state behavior
+    waitTillAllSlotsAreOccupied(true);
+    waitTillAllSlotsAreOccupied(false);
+    numTasks = 1;
+    for (int i = 0; i < 4; i++) {
+      ControlledMapReduceJob.assertNumTasksRunning(jips[i], true, numTasks);
+      ControlledMapReduceJob.assertNumTasksRunning(jips[i], false, numTasks);
+      numTasks += 1;
+    }
+
+    // Finish the jobs and join them
+    numTasks = 1;
+    for (int i = 0; i < 4; i++) {
+      controlledJobs[i].finishJob();
+      ControlledMapReduceJob
+          .waitTillNTotalTasksFinish(jips[i], true, numTasks);
+      ControlledMapReduceJob.assertNumTasksRunning(jips[i], true, 0);
+      ControlledMapReduceJob.waitTillNTotalTasksFinish(jips[i], false,
+          numTasks);
+      ControlledMapReduceJob.assertNumTasksRunning(jips[i], false, 0);
+      jobRunners[i].join();
+      numTasks += 1;
+    }
+  }
+}

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

@@ -154,6 +154,10 @@ public abstract class ClusterMapReduceTestCase extends TestCase {
     return dfsCluster.getFileSystem();
   }
 
+  protected MiniMRCluster getMRCluster() {
+    return mrCluster;
+  }
+
   /**
    * Returns the path to the root directory for the testcase.
    *

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

@@ -0,0 +1,576 @@
+/**
+ * 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.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+import org.apache.hadoop.mapred.lib.NullOutputFormat;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+
+/**
+ * A Controlled Map/Reduce Job. The tasks are controlled by the presence of
+ * particularly named files in the directory signalFileDir on the file-system
+ * that the job is configured to work with. Tasks get scheduled by the
+ * scheduler, occupy the slots on the TaskTrackers and keep running till the
+ * user gives a signal via files whose names are of the form MAPS_[0-9]* and
+ * REDUCES_[0-9]*. For e.g., whenever the map tasks see that a file name MAPS_5
+ * is created in the singalFileDir, all the maps whose TaskAttemptIDs are below
+ * 4 get finished. At any time, there should be only one MAPS_[0-9]* file and
+ * only one REDUCES_[0-9]* file in the singnalFileDir. In the beginning MAPS_0
+ * and REDUCE_0 files are present, and further signals are given by renaming
+ * these files.
+ * 
+ */
+class ControlledMapReduceJob extends Configured implements Tool,
+    Mapper<NullWritable, NullWritable, IntWritable, NullWritable>,
+    Reducer<IntWritable, NullWritable, NullWritable, NullWritable>,
+    Partitioner<IntWritable, NullWritable>,
+    InputFormat<NullWritable, NullWritable> {
+
+  static final Log LOG = LogFactory.getLog(ControlledMapReduceJob.class);
+
+  private FileSystem fs = null;
+  private int taskNumber;
+
+  private static ArrayList<Path> signalFileDirCache = new ArrayList<Path>();
+
+  private Path signalFileDir;
+  {
+    Random random = new Random();
+    signalFileDir = new Path("signalFileDir-" + random.nextLong());
+    while (signalFileDirCache.contains(signalFileDir)) {
+      signalFileDir = new Path("signalFileDir-" + random.nextLong());
+    }
+    signalFileDirCache.add(signalFileDir);
+  }
+
+  private long mapsFinished = 0;
+  private long reducesFinished = 0;
+
+  private RunningJob rJob = null;
+
+  private int numMappers;
+  private int numReducers;
+
+  private final String MAP_SIGFILE_PREFIX = "MAPS_";
+  private final String REDUCE_SIGFILE_PREFIX = "REDUCES_";
+
+  private void initialize()
+      throws IOException {
+    fs = FileSystem.get(getConf());
+    fs.mkdirs(signalFileDir);
+    writeFile(new Path(signalFileDir, MAP_SIGFILE_PREFIX + mapsFinished));
+    writeFile(new Path(signalFileDir, REDUCE_SIGFILE_PREFIX + reducesFinished));
+  }
+
+  /**
+   * Finish N number of maps/reduces.
+   * 
+   * @param isMap
+   * @param noOfTasksToFinish
+   * @throws IOException
+   */
+  public void finishNTasks(boolean isMap, int noOfTasksToFinish)
+      throws IOException {
+    if (noOfTasksToFinish < 0) {
+      throw new IOException(
+          "Negative values for noOfTasksToFinish not acceptable");
+    }
+
+    if (noOfTasksToFinish == 0) {
+      return;
+    }
+
+    LOG.info("Going to finish off " + noOfTasksToFinish);
+    String PREFIX = isMap ? MAP_SIGFILE_PREFIX : REDUCE_SIGFILE_PREFIX;
+    long tasksFinished = isMap ? mapsFinished : reducesFinished;
+    Path oldSignalFile =
+        new Path(signalFileDir, PREFIX + String.valueOf(tasksFinished));
+    Path newSignalFile =
+        new Path(signalFileDir, PREFIX
+            + String.valueOf(tasksFinished + noOfTasksToFinish));
+    fs.rename(oldSignalFile, newSignalFile);
+    if (isMap) {
+      mapsFinished += noOfTasksToFinish;
+    } else {
+      reducesFinished += noOfTasksToFinish;
+    }
+    LOG.info("Successfully sent signal to finish off " + noOfTasksToFinish);
+  }
+
+  /**
+   * Finished all tasks of type determined by isMap
+   * 
+   * @param isMap
+   * @throws IOException
+   */
+  public void finishAllTasks(boolean isMap)
+      throws IOException {
+    finishNTasks(isMap, (isMap ? numMappers : numReducers));
+  }
+
+  /**
+   * Finish the job
+   * 
+   * @throws IOException
+   */
+  public void finishJob()
+      throws IOException {
+    finishAllTasks(true);
+    finishAllTasks(false);
+  }
+
+  /**
+   * Wait till noOfTasksToBeRunning number of tasks of type specified by isMap
+   * started running. This currently uses a jip object and directly uses its api
+   * to determine the number of tasks running.
+   * 
+   * <p>
+   * 
+   * TODO: It should eventually use a JobID and then get the information from
+   * the JT to check the number of running tasks.
+   * 
+   * @param jip
+   * @param isMap
+   * @param noOfTasksToBeRunning
+   */
+  static void waitTillNTasksStartRunning(JobInProgress jip, boolean isMap,
+      int noOfTasksToBeRunning)
+      throws InterruptedException {
+    int numTasks = 0;
+    while (numTasks != noOfTasksToBeRunning) {
+      Thread.sleep(1000);
+      numTasks = isMap ? jip.runningMaps() : jip.runningReduces();
+      LOG.info("Waiting till " + noOfTasksToBeRunning
+          + (isMap ? " map" : " reduce") + " tasks of the job "
+          + jip.getJobID() + " start running. " + numTasks
+          + " tasks already started running.");
+    }
+  }
+
+  /**
+   * Make sure that the number of tasks of type specified by isMap running in
+   * the given job is the same as noOfTasksToBeRunning
+   * 
+   * <p>
+   * 
+   * TODO: It should eventually use a JobID and then get the information from
+   * the JT to check the number of running tasks.
+   * 
+   * @param jip
+   * @param isMap
+   * @param noOfTasksToBeRunning
+   */
+  static void assertNumTasksRunning(JobInProgress jip, boolean isMap,
+      int noOfTasksToBeRunning)
+      throws Exception {
+    if ((isMap ? jip.runningMaps() : jip.runningReduces()) != noOfTasksToBeRunning) {
+      throw new Exception("Number of tasks running is not "
+          + noOfTasksToBeRunning);
+    }
+  }
+
+  /**
+   * Wait till noOfTasksToFinish number of tasks of type specified by isMap
+   * are finished. This currently uses a jip object and directly uses its api to
+   * determine the number of tasks finished.
+   * 
+   * <p>
+   * 
+   * TODO: It should eventually use a JobID and then get the information from
+   * the JT to check the number of finished tasks.
+   * 
+   * @param jip
+   * @param isMap
+   * @param noOfTasksToFinish
+   * @throws InterruptedException
+   */
+  static void waitTillNTotalTasksFinish(JobInProgress jip, boolean isMap,
+      int noOfTasksToFinish)
+      throws InterruptedException {
+    int noOfTasksAlreadyFinished = 0;
+    while (noOfTasksAlreadyFinished < noOfTasksToFinish) {
+      Thread.sleep(1000);
+      noOfTasksAlreadyFinished =
+          (isMap ? jip.finishedMaps() : jip.finishedReduces());
+      LOG.info("Waiting till " + noOfTasksToFinish
+          + (isMap ? " map" : " reduce") + " tasks of the job "
+          + jip.getJobID() + " finish. " + noOfTasksAlreadyFinished
+          + " tasks already got finished.");
+    }
+  }
+
+  /**
+   * Have all the tasks of type specified by isMap finished in this job?
+   * 
+   * @param jip
+   * @param isMap
+   * @return true if finished, false otherwise
+   */
+  static boolean haveAllTasksFinished(JobInProgress jip, boolean isMap) {
+    return ((isMap ? jip.runningMaps() : jip.runningReduces()) == 0);
+  }
+
+  private void writeFile(Path name)
+      throws IOException {
+    Configuration conf = new Configuration(false);
+    SequenceFile.Writer writer =
+        SequenceFile.createWriter(fs, conf, name, BytesWritable.class,
+            BytesWritable.class, CompressionType.NONE);
+    writer.append(new BytesWritable(), new BytesWritable());
+    writer.close();
+  }
+
+  @Override
+  public void configure(JobConf conf) {
+    try {
+      signalFileDir = new Path(conf.get("signal.dir.path"));
+      numReducers = conf.getNumReduceTasks();
+      fs = FileSystem.get(conf);
+      String taskAttemptId = conf.get("mapred.task.id");
+      if (taskAttemptId != null) {
+        TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskAttemptId);
+        taskNumber = taskAttemptID.getTaskID().getId();
+      }
+    } catch (IOException ioe) {
+      LOG.warn("Caught exception " + ioe);
+    }
+  }
+
+  private FileStatus[] listSignalFiles(FileSystem fileSys, final boolean isMap)
+      throws IOException {
+    return fileSys.globStatus(new Path(signalFileDir.toString() + "/*"),
+        new PathFilter() {
+          @Override
+          public boolean accept(Path path) {
+            if (isMap && path.getName().startsWith(MAP_SIGFILE_PREFIX)) {
+              LOG.debug("Found signal file : " + path.getName());
+              return true;
+            } else if (!isMap
+                && path.getName().startsWith(REDUCE_SIGFILE_PREFIX)) {
+              LOG.debug("Found signal file : " + path.getName());
+              return true;
+            }
+            LOG.info("Didn't find any relevant signal files.");
+            return false;
+          }
+        });
+  }
+
+  @Override
+  public void map(NullWritable key, NullWritable value,
+      OutputCollector<IntWritable, NullWritable> output, Reporter reporter)
+      throws IOException {
+    LOG.info(taskNumber + " has started.");
+    FileStatus[] files = listSignalFiles(fs, true);
+    String[] sigFileComps = files[0].getPath().getName().split("_");
+    String signalType = sigFileComps[0];
+    int noOfTasks = Integer.parseInt(sigFileComps[1]);
+
+    while (!signalType.equals("MAPS") || taskNumber + 1 > noOfTasks) {
+      LOG.info("Signal type found : " + signalType
+          + " .Number of tasks to be finished by this signal : " + noOfTasks
+          + " . My id : " + taskNumber);
+      LOG.info(taskNumber + " is still alive.");
+      try {
+        reporter.progress();
+        Thread.sleep(1000);
+      } catch (InterruptedException ie) {
+        LOG.info(taskNumber + " is still alive.");
+        break;
+      }
+      files = listSignalFiles(fs, true);
+      sigFileComps = files[0].getPath().getName().split("_");
+      signalType = sigFileComps[0];
+      noOfTasks = Integer.parseInt(sigFileComps[1]);
+    }
+    LOG.info("Signal type found : " + signalType
+        + " .Number of tasks to be finished by this signal : " + noOfTasks
+        + " . My id : " + taskNumber);
+    // output numReduce number of random values, so that
+    // each reducer will get one key each.
+    for (int i = 0; i < numReducers; i++) {
+      output.collect(new IntWritable(i), NullWritable.get());
+    }
+
+    LOG.info(taskNumber + " is finished.");
+  }
+
+  @Override
+  public void reduce(IntWritable key, Iterator<NullWritable> values,
+      OutputCollector<NullWritable, NullWritable> output, Reporter reporter)
+      throws IOException {
+    LOG.info(taskNumber + " has started.");
+    FileStatus[] files = listSignalFiles(fs, false);
+    String[] sigFileComps = files[0].getPath().getName().split("_");
+    String signalType = sigFileComps[0];
+    int noOfTasks = Integer.parseInt(sigFileComps[1]);
+
+    while (!signalType.equals("REDUCES") || taskNumber + 1 > noOfTasks) {
+      LOG.info("Signal type found : " + signalType
+          + " .Number of tasks to be finished by this signal : " + noOfTasks
+          + " . My id : " + taskNumber);
+      LOG.info(taskNumber + " is still alive.");
+      try {
+        reporter.progress();
+        Thread.sleep(1000);
+      } catch (InterruptedException ie) {
+        LOG.info(taskNumber + " is still alive.");
+        break;
+      }
+      files = listSignalFiles(fs, false);
+      sigFileComps = files[0].getPath().getName().split("_");
+      signalType = sigFileComps[0];
+      noOfTasks = Integer.parseInt(sigFileComps[1]);
+    }
+    LOG.info("Signal type found : " + signalType
+        + " .Number of tasks to be finished by this signal : " + noOfTasks
+        + " . My id : " + taskNumber);
+    LOG.info(taskNumber + " is finished.");
+  }
+
+  @Override
+  public void close()
+      throws IOException {
+    // nothing
+  }
+
+  public JobID getJobId() {
+    if (rJob == null) {
+      return null;
+    }
+    return rJob.getID();
+  }
+
+  public int run(int numMapper, int numReducer)
+      throws IOException {
+    JobConf conf =
+        getControlledMapReduceJobConf(getConf(), numMapper, numReducer);
+    JobClient client = new JobClient(conf);
+    rJob = client.submitJob(conf);
+    while (!rJob.isComplete()) {
+      try {
+        Thread.sleep(1000);
+      } catch (InterruptedException ie) {
+        break;
+      }
+    }
+    if (rJob.isSuccessful()) {
+      return 0;
+    }
+    return 1;
+  }
+
+  private JobConf getControlledMapReduceJobConf(Configuration clusterConf,
+      int numMapper, int numReducer)
+      throws IOException {
+    setConf(clusterConf);
+    initialize();
+    JobConf conf = new JobConf(getConf(), ControlledMapReduceJob.class);
+    conf.setJobName("ControlledJob");
+    conf.set("signal.dir.path", signalFileDir.toString());
+    conf.setNumMapTasks(numMapper);
+    conf.setNumReduceTasks(numReducer);
+    conf.setMapperClass(ControlledMapReduceJob.class);
+    conf.setMapOutputKeyClass(IntWritable.class);
+    conf.setMapOutputValueClass(NullWritable.class);
+    conf.setReducerClass(ControlledMapReduceJob.class);
+    conf.setOutputKeyClass(NullWritable.class);
+    conf.setOutputValueClass(NullWritable.class);
+    conf.setInputFormat(ControlledMapReduceJob.class);
+    FileInputFormat.addInputPath(conf, new Path("ignored"));
+    conf.setOutputFormat(NullOutputFormat.class);
+
+    // Set the following for reduce tasks to be able to be started running
+    // immediately along with maps.
+    conf.set("mapred.reduce.slowstart.completed.maps", String.valueOf(0));
+
+    return conf;
+  }
+
+  @Override
+  public int run(String[] args)
+      throws Exception {
+    numMappers = Integer.parseInt(args[0]);
+    numReducers = Integer.parseInt(args[1]);
+    return run(numMappers, numReducers);
+  }
+
+  @Override
+  public int getPartition(IntWritable k, NullWritable v, int numPartitions) {
+    return k.get() % numPartitions;
+  }
+
+  @Override
+  public RecordReader<NullWritable, NullWritable> getRecordReader(
+      InputSplit split, JobConf job, Reporter reporter) {
+    LOG.debug("Inside RecordReader.getRecordReader");
+    return new RecordReader<NullWritable, NullWritable>() {
+      private int pos = 0;
+
+      public void close() {
+        // nothing
+      }
+
+      public NullWritable createKey() {
+        return NullWritable.get();
+      }
+
+      public NullWritable createValue() {
+        return NullWritable.get();
+      }
+
+      public long getPos() {
+        return pos;
+      }
+
+      public float getProgress() {
+        return pos * 100;
+      }
+
+      public boolean next(NullWritable key, NullWritable value) {
+        if (pos++ == 0) {
+          LOG.debug("Returning the next record");
+          return true;
+        }
+        LOG.debug("No more records. Returning none.");
+        return false;
+      }
+    };
+  }
+
+  @Override
+  public InputSplit[] getSplits(JobConf job, int numSplits) {
+    LOG.debug("Inside InputSplit.getSplits");
+    InputSplit[] ret = new InputSplit[numSplits];
+    for (int i = 0; i < numSplits; ++i) {
+      ret[i] = new EmptySplit();
+    }
+    return ret;
+  }
+
+  public static class EmptySplit implements InputSplit {
+    public void write(DataOutput out)
+        throws IOException {
+    }
+
+    public void readFields(DataInput in)
+        throws IOException {
+    }
+
+    public long getLength() {
+      return 0L;
+    }
+
+    public String[] getLocations() {
+      return new String[0];
+    }
+  }
+
+  static class ControlledMapReduceJobRunner extends Thread {
+    private JobConf conf;
+    private ControlledMapReduceJob job;
+    private JobID jobID;
+
+    private int numMappers;
+    private int numReducers;
+
+    public ControlledMapReduceJobRunner() {
+      this(new JobConf(), 5, 5);
+    }
+
+    public ControlledMapReduceJobRunner(JobConf cnf, int numMap, int numRed) {
+      this.conf = cnf;
+      this.numMappers = numMap;
+      this.numReducers = numRed;
+    }
+
+    public ControlledMapReduceJob getJob() {
+      while (job == null) {
+        try {
+          Thread.sleep(1000);
+        } catch (InterruptedException ie) {
+          LOG.info(ControlledMapReduceJobRunner.class.getName()
+              + " is interrupted.");
+          break;
+        }
+      }
+      return job;
+    }
+
+    public JobID getJobID()
+        throws IOException {
+      ControlledMapReduceJob job = getJob();
+      JobID id = job.getJobId();
+      while (id == null) {
+        id = job.getJobId();
+        try {
+          Thread.sleep(1000);
+        } catch (InterruptedException ie) {
+          LOG.info(ControlledMapReduceJobRunner.class.getName()
+              + " is interrupted.");
+          break;
+        }
+      }
+      return id;
+    }
+
+    @Override
+    public void run() {
+      if (job != null) {
+        LOG.warn("Job is already running.");
+        return;
+      }
+      try {
+        job = new ControlledMapReduceJob();
+        int ret =
+            ToolRunner.run(this.conf, job, new String[] {
+                String.valueOf(numMappers), String.valueOf(numReducers) });
+        LOG.info("Return value for the job : " + ret);
+      } catch (Exception e) {
+        LOG.warn("Caught exception : " + StringUtils.stringifyException(e));
+      }
+    }
+
+    static ControlledMapReduceJobRunner getControlledMapReduceJobRunner(
+        JobConf conf, int numMappers, int numReducers) {
+      return new ControlledMapReduceJobRunner(conf, numMappers, numReducers);
+    }
+  }
+}

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

@@ -0,0 +1,81 @@
+/**
+ * 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.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.ControlledMapReduceJob.ControlledMapReduceJobRunner;
+
+/**
+ * Test to verify the controlled behavior of a ControlledMapReduceJob.
+ * 
+ */
+public class TestControlledMapReduceJob extends ClusterMapReduceTestCase {
+  static final Log LOG = LogFactory.getLog(TestControlledMapReduceJob.class);
+
+  /**
+   * Starts a job with 5 maps and 5 reduces. Then controls the finishing of
+   * tasks. Signals finishing tasks in batches and then verifies their
+   * completion.
+   * 
+   * @throws Exception
+   */
+  public void testControlledMapReduceJob()
+      throws Exception {
+
+    Properties props = new Properties();
+    props.setProperty("mapred.tasktracker.map.tasks.maximum", "2");
+    props.setProperty("mapred.tasktracker.reduce.tasks.maximum", "2");
+    startCluster(true, props);
+    LOG.info("Started the cluster");
+
+    ControlledMapReduceJobRunner jobRunner =
+        ControlledMapReduceJobRunner
+            .getControlledMapReduceJobRunner(createJobConf(), 7, 6);
+    jobRunner.start();
+    ControlledMapReduceJob controlledJob = jobRunner.getJob();
+    JobInProgress jip =
+        getMRCluster().getJobTrackerRunner().getJobTracker().getJob(
+            jobRunner.getJobID());
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip, true, 4);
+    LOG.info("Finishing 3 maps");
+    controlledJob.finishNTasks(true, 3);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip, true, 3);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip, true, 4);
+    LOG.info("Finishing 4 more maps");
+    controlledJob.finishNTasks(true, 4);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip, true, 7);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip, false, 4);
+    LOG.info("Finishing 2 reduces");
+    controlledJob.finishNTasks(false, 2);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip, false, 2);
+
+    ControlledMapReduceJob.waitTillNTasksStartRunning(jip, false, 4);
+    LOG.info("Finishing 4 more reduces");
+    controlledJob.finishNTasks(false, 4);
+    ControlledMapReduceJob.waitTillNTotalTasksFinish(jip, false, 6);
+
+    jobRunner.join();
+  }
+}