mapreduce/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestRefreshOfQueues.java - hadoop - Git at Google

 /**
  * 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 static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;

 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.CapacityTestUtils.ControlledInitializationPoller;
 import org.apache.hadoop.mapred.CapacityTestUtils.FakeJobInProgress;
 import org.apache.hadoop.mapred.CapacityTestUtils.FakeTaskTrackerManager;
 import static org.apache.hadoop.mapred.CapacityTestUtils.*;
 import org.junit.After;
 import org.junit.Test;

 /**
  * Test the Queue-Refresh feature.
  */
 public class TestRefreshOfQueues {

   private static final Log LOG =
       LogFactory.getLog(org.apache.hadoop.mapred.TestRefreshOfQueues.class);

   String queueConfigPath =
       System.getProperty("test.build.extraconf", "build/test/extraconf");
   File queueConfigFile =
       new File(queueConfigPath, QueueManager.QUEUE_CONF_FILE_NAME);

   private CapacityTaskScheduler scheduler;
   private FakeTaskTrackerManager taskTrackerManager;

   /**
    * Remove the queueConfigFile once the test is done.
    */
   @After
   public void tearDown() {
     if (queueConfigFile.exists()) {
       queueConfigFile.delete();
     }
   }

   /**
    * Sets up the scheduler, TaskTrackerManager, QueueManager, initializer and
    * starts the scheduler.
    *
    * @throws IOException
    */
   private void setupAndStartSchedulerFramework(int numTTs, int numMapsPerTT,
       int numReducesPerTT)
       throws IOException {
     scheduler = new CapacityTaskScheduler();
     taskTrackerManager =
         new FakeTaskTrackerManager(numTTs, numMapsPerTT, numReducesPerTT);
     taskTrackerManager.setQueueManager(new QueueManager());
     scheduler.setTaskTrackerManager(taskTrackerManager);
     scheduler.setConf(new Configuration());
     ControlledInitializationPoller controlledInitializationPoller =
         new ControlledInitializationPoller(scheduler.jobQueuesManager,
             taskTrackerManager);
     scheduler.setInitializationPoller(controlledInitializationPoller);
     taskTrackerManager.addJobInProgressListener(scheduler.jobQueuesManager);
     scheduler.start();
   }

   /**
    * Helper method that ensures TaskScheduler is locked before calling
    * {@link QueueManager#refreshQueues(Configuration,
    *    org.apache.hadoop.mapred.TaskScheduler.QueueRefresher)}.
    */
   private static void refreshQueues(QueueManager qm, Configuration conf,
       TaskScheduler ts) throws IOException {
     synchronized (ts) {
       qm.refreshQueues(conf, ts.getQueueRefresher());
     }
   }

   /**
    * @throws Throwable
    */
   @Test
   public void testRefreshOfQueuesSanity()
       throws Throwable {

     JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

     Properties[] props = new Properties[3];
     for (int i = 0; i < props.length; i++) {
       props[i] = queues[i].getProperties();
       props[i].setProperty(CapacitySchedulerConf.CAPACITY_PROPERTY,
           String.valueOf(i + 10));
       props[i].setProperty(CapacitySchedulerConf.MAX_CAPACITY_PROPERTY,
           String.valueOf(i + 15));
       props[i].setProperty(CapacitySchedulerConf.SUPPORTS_PRIORITY_PROPERTY,
           String.valueOf(false));
       props[i].setProperty(
           CapacitySchedulerConf.MAXIMUM_INITIALIZED_JOBS_PER_USER_PROPERTY,
           String.valueOf(i + 11));
       props[i].setProperty(
           CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
           String.valueOf(i + 16));
     }

     // write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
         queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

     setupAndStartSchedulerFramework(0, 0, 0);

     Map<String, AbstractQueue> allQueues = getAllQueues(scheduler);

     // Verify the configuration.
     for (int i = 0; i < queues.length; i++) {
       String qName = queues[i].getQueueName();
       LOG.info("Queue name : " + qName);
       QueueSchedulingContext qsc =
           allQueues.get(qName).getQueueSchedulingContext();
       LOG.info("Context for queue " + qName + " is : " + qsc);
       assertEquals(i + 10, qsc.getCapacityPercent(), 0);
       assertEquals(i + 15, qsc.getMaxCapacityPercent(), 0);
       assertEquals(Boolean.valueOf(false),
           Boolean.valueOf(qsc.supportsPriorities()));
       assertEquals(i + 16, qsc.getUlMin());
     }

     // change configuration
     for (int i = 0; i < props.length; i++) {
       props[i] = queues[i].getProperties();
       props[i].setProperty(CapacitySchedulerConf.CAPACITY_PROPERTY,
           String.valueOf(i + 20));
       props[i].setProperty(CapacitySchedulerConf.MAX_CAPACITY_PROPERTY,
           String.valueOf(i + 25));
       props[i].setProperty(CapacitySchedulerConf.SUPPORTS_PRIORITY_PROPERTY,
           String.valueOf(false));
       props[i].setProperty(
           CapacitySchedulerConf.MAXIMUM_INITIALIZED_JOBS_PER_USER_PROPERTY,
           String.valueOf(i + 5));
       props[i].setProperty(
           CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
           String.valueOf(i + 10));
     }

     // Re-write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
         queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

     // Now do scheduler refresh.
     refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);

     allQueues = getAllQueues(scheduler);

     for (int i = 0; i < queues.length; i++) {
       String qName = queues[i].getQueueName();
       LOG.info("Queue name : " + qName);
       QueueSchedulingContext qsc =
         allQueues.get(qName).getQueueSchedulingContext();
       assertEquals(qName, qsc.getQueueName());
       LOG.info("Context for queue " + qName + " is : " + qsc);
       assertEquals(i + 20, qsc.getCapacityPercent(), 0);
       assertEquals(i + 25, qsc.getMaxCapacityPercent(), 0);
       assertEquals(Boolean.valueOf(false),
           Boolean.valueOf(qsc.supportsPriorities()));
     }
   }

   /**
    * @throws Throwable
    */
    @Test
   public void testSuccessfulCapacityRefresh()
       throws Throwable {

     JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

     queues[0].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(100));
     queues[1].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));
     queues[2].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

     // write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
       queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

     setupAndStartSchedulerFramework(2, 2, 2);

     FakeJobInProgress job1 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[1].getQueueName(), "user");
     FakeJobInProgress job2 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[2].getQueueName(), "user");

     Map<String, String> expectedStrings = new HashMap<String, String>();
     expectedStrings.put(MAP, "attempt_test_0001_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0001_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);
 //===========================================
     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0002_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0002_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);
 //============================================
     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0002_m_000002_0 on tt2");
     expectedStrings.put(REDUCE, "attempt_test_0002_r_000002_0 on tt2");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt2",
       expectedStrings);
 //============================================
     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0001_m_000002_0 on tt2");
     expectedStrings.put(REDUCE, "attempt_test_0001_r_000002_0 on tt2");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt2",
       expectedStrings);

     taskTrackerManager.killJob(job1.getJobID());
     taskTrackerManager.killJob(job2.getJobID());

     // change configuration
     queues[1].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(25));
     queues[2].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(75));

     // Re-write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
       queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

     refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);

     job1 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[1].getQueueName(), "user");
     job2 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 4, 4,
         queues[2].getQueueName(), "user");

     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0003_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0003_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);


     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0004_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0004_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);


     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0004_m_000002_0 on tt2");
     expectedStrings.put(REDUCE, "attempt_test_0004_r_000002_0 on tt2");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt2",
       expectedStrings);

     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0004_m_000003_0 on tt2");
     expectedStrings.put(REDUCE, "attempt_test_0004_r_000003_0 on tt2");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt2",
       expectedStrings);

   }

   /**
    * Test to verify that the refresh of the scheduler fails when modified
    * configuration overflows 100%
    *
    * @throws Throwable
    */
    @Test
   public void testFailingCapacityRefresh()
       throws Throwable {

     JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

     queues[0].getProperties().setProperty(
         CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(100));
     queues[1].getProperties().setProperty(
         CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(70));
     queues[2].getProperties().setProperty(
         CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

     // write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
         queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

     try {
       setupAndStartSchedulerFramework(2, 2, 2);
       fail("Scheduler should have failed to start!");
     } catch (IOException ioe) {
       assertTrue(ioe.getMessage().contains(
           String.format(QueueHierarchyBuilder.TOTAL_CAPACITY_OVERFLOWN_MSG,
               queues[1].getQueueName() + "," + queues[2].getQueueName(),
               Float.valueOf(120.0f))));
     }

     // Rectify the properties and start the scheduler
     queues[1].getProperties().setProperty(
         CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

     // write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
         queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

     setupAndStartSchedulerFramework(2, 2, 2);

     // Now change configuration.
     queues[1].getProperties().setProperty(
         CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(35));
     queues[2].getProperties().setProperty(
         CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(95));

     // Re-write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
         queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

     try {
       refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);
     } catch (IOException ioe) {
       assertTrue(ioe.getMessage().contains(
           String.format(QueueHierarchyBuilder.TOTAL_CAPACITY_OVERFLOWN_MSG,
               queues[1].getQueueName() + "," + queues[2].getQueueName(),
               Float.valueOf(130.0f))));
     }
   }

   /**
    * @throws Throwable
    */
    @Test
   public void testRefreshUserLimits()
       throws Throwable {

     JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

     queues[0].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(100));
     queues[1].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));
     queues[2].getProperties().setProperty(
       CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

     queues[2].getProperties().setProperty(
       CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
       String.valueOf(100));

     // write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
       queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

     setupAndStartSchedulerFramework(1, 2, 2);

     FakeJobInProgress job1 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[2].getQueueName(), "user1");
     FakeJobInProgress job2 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[2].getQueueName(), "user2");

     Map<String, String> expectedStrings = new HashMap<String, String>();
     expectedStrings.put(MAP, "attempt_test_0001_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0001_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);

     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0001_m_000002_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0001_r_000002_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);

     assertNull(scheduler.assignTasks(taskTrackerManager.getTaskTracker("tt1")));
     taskTrackerManager.killJob(job1.getJobID());
     taskTrackerManager.killJob(job2.getJobID());

     // change configuration
     queues[2].getProperties().setProperty(
       CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
       String.valueOf(50));

     // Re-write the configuration file
     QueueManagerTestUtils.writeQueueConfigurationFile(
       queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

     refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);

     job1 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[1].getQueueName(), "user1");
     job2 =
       taskTrackerManager.submitJobAndInit(
         JobStatus.PREP, 2, 2,
         queues[2].getQueueName(), "user2");

     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0003_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0003_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);

     expectedStrings.clear();
     expectedStrings.put(MAP, "attempt_test_0004_m_000001_0 on tt1");
     expectedStrings.put(REDUCE, "attempt_test_0004_r_000001_0 on tt1");
     checkMultipleTaskAssignment(
       taskTrackerManager, scheduler, "tt1",
       expectedStrings);
   }

   /**
    * Get a map of all {@link AbstractQueue}s.
    *
    * @param sched
    * @return
    */
   private static Map<String, AbstractQueue> getAllQueues(
       CapacityTaskScheduler sched) {
     AbstractQueue rootQueue = sched.getRoot();
     HashMap<String, AbstractQueue> allQueues =
         new HashMap<String, AbstractQueue>();
     List<AbstractQueue> allQueuesList = new ArrayList<AbstractQueue>();
     allQueuesList.addAll(rootQueue.getDescendentJobQueues());
     allQueuesList.addAll(rootQueue.getDescendantContainerQueues());
     for (AbstractQueue q : allQueuesList) {
       LOG.info("Putting in allQueues list " + q.getName());
       allQueues.put(q.getName(), q);
     }
     return allQueues;
   }
 }
	/**
	* 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 static org.junit.Assert.assertTrue;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNull;
	import static org.junit.Assert.fail;

	import java.io.File;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Properties;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.mapred.CapacityTestUtils.ControlledInitializationPoller;
	import org.apache.hadoop.mapred.CapacityTestUtils.FakeJobInProgress;
	import org.apache.hadoop.mapred.CapacityTestUtils.FakeTaskTrackerManager;
	import static org.apache.hadoop.mapred.CapacityTestUtils.*;
	import org.junit.After;
	import org.junit.Test;

	/**
	* Test the Queue-Refresh feature.
	*/
	public class TestRefreshOfQueues {

	private static final Log LOG =
	LogFactory.getLog(org.apache.hadoop.mapred.TestRefreshOfQueues.class);

	String queueConfigPath =
	System.getProperty("test.build.extraconf", "build/test/extraconf");
	File queueConfigFile =
	new File(queueConfigPath, QueueManager.QUEUE_CONF_FILE_NAME);

	private CapacityTaskScheduler scheduler;
	private FakeTaskTrackerManager taskTrackerManager;

	/**
	* Remove the queueConfigFile once the test is done.
	*/
	@After
	public void tearDown() {
	if (queueConfigFile.exists()) {
	queueConfigFile.delete();
	}
	}

	/**
	* Sets up the scheduler, TaskTrackerManager, QueueManager, initializer and
	* starts the scheduler.
	*
	* @throws IOException
	*/
	private void setupAndStartSchedulerFramework(int numTTs, int numMapsPerTT,
	int numReducesPerTT)
	throws IOException {
	scheduler = new CapacityTaskScheduler();
	taskTrackerManager =
	new FakeTaskTrackerManager(numTTs, numMapsPerTT, numReducesPerTT);
	taskTrackerManager.setQueueManager(new QueueManager());
	scheduler.setTaskTrackerManager(taskTrackerManager);
	scheduler.setConf(new Configuration());
	ControlledInitializationPoller controlledInitializationPoller =
	new ControlledInitializationPoller(scheduler.jobQueuesManager,
	taskTrackerManager);
	scheduler.setInitializationPoller(controlledInitializationPoller);
	taskTrackerManager.addJobInProgressListener(scheduler.jobQueuesManager);
	scheduler.start();
	}

	/**
	* Helper method that ensures TaskScheduler is locked before calling
	* {@link QueueManager#refreshQueues(Configuration,
	* org.apache.hadoop.mapred.TaskScheduler.QueueRefresher)}.
	*/
	private static void refreshQueues(QueueManager qm, Configuration conf,
	TaskScheduler ts) throws IOException {
	synchronized (ts) {
	qm.refreshQueues(conf, ts.getQueueRefresher());
	}
	}

	/**
	* @throws Throwable
	*/
	@Test
	public void testRefreshOfQueuesSanity()
	throws Throwable {

	JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

	Properties[] props = new Properties[3];
	for (int i = 0; i < props.length; i++) {
	props[i] = queues[i].getProperties();
	props[i].setProperty(CapacitySchedulerConf.CAPACITY_PROPERTY,
	String.valueOf(i + 10));
	props[i].setProperty(CapacitySchedulerConf.MAX_CAPACITY_PROPERTY,
	String.valueOf(i + 15));
	props[i].setProperty(CapacitySchedulerConf.SUPPORTS_PRIORITY_PROPERTY,
	String.valueOf(false));
	props[i].setProperty(
	CapacitySchedulerConf.MAXIMUM_INITIALIZED_JOBS_PER_USER_PROPERTY,
	String.valueOf(i + 11));
	props[i].setProperty(
	CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
	String.valueOf(i + 16));
	}

	// write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

	setupAndStartSchedulerFramework(0, 0, 0);

	Map<String, AbstractQueue> allQueues = getAllQueues(scheduler);

	// Verify the configuration.
	for (int i = 0; i < queues.length; i++) {
	String qName = queues[i].getQueueName();
	LOG.info("Queue name : " + qName);
	QueueSchedulingContext qsc =
	allQueues.get(qName).getQueueSchedulingContext();
	LOG.info("Context for queue " + qName + " is : " + qsc);
	assertEquals(i + 10, qsc.getCapacityPercent(), 0);
	assertEquals(i + 15, qsc.getMaxCapacityPercent(), 0);
	assertEquals(Boolean.valueOf(false),
	Boolean.valueOf(qsc.supportsPriorities()));
	assertEquals(i + 16, qsc.getUlMin());
	}

	// change configuration
	for (int i = 0; i < props.length; i++) {
	props[i] = queues[i].getProperties();
	props[i].setProperty(CapacitySchedulerConf.CAPACITY_PROPERTY,
	String.valueOf(i + 20));
	props[i].setProperty(CapacitySchedulerConf.MAX_CAPACITY_PROPERTY,
	String.valueOf(i + 25));
	props[i].setProperty(CapacitySchedulerConf.SUPPORTS_PRIORITY_PROPERTY,
	String.valueOf(false));
	props[i].setProperty(
	CapacitySchedulerConf.MAXIMUM_INITIALIZED_JOBS_PER_USER_PROPERTY,
	String.valueOf(i + 5));
	props[i].setProperty(
	CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
	String.valueOf(i + 10));
	}

	// Re-write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

	// Now do scheduler refresh.
	refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);

	allQueues = getAllQueues(scheduler);

	for (int i = 0; i < queues.length; i++) {
	String qName = queues[i].getQueueName();
	LOG.info("Queue name : " + qName);
	QueueSchedulingContext qsc =
	allQueues.get(qName).getQueueSchedulingContext();
	assertEquals(qName, qsc.getQueueName());
	LOG.info("Context for queue " + qName + " is : " + qsc);
	assertEquals(i + 20, qsc.getCapacityPercent(), 0);
	assertEquals(i + 25, qsc.getMaxCapacityPercent(), 0);
	assertEquals(Boolean.valueOf(false),
	Boolean.valueOf(qsc.supportsPriorities()));
	}
	}

	/**
	* @throws Throwable
	*/
	@Test
	public void testSuccessfulCapacityRefresh()
	throws Throwable {

	JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

	queues[0].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(100));
	queues[1].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));
	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

	// write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

	setupAndStartSchedulerFramework(2, 2, 2);

	FakeJobInProgress job1 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[1].getQueueName(), "user");
	FakeJobInProgress job2 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[2].getQueueName(), "user");

	Map<String, String> expectedStrings = new HashMap<String, String>();
	expectedStrings.put(MAP, "attempt_test_0001_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0001_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);
	//===========================================
	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0002_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0002_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);
	//============================================
	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0002_m_000002_0 on tt2");
	expectedStrings.put(REDUCE, "attempt_test_0002_r_000002_0 on tt2");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt2",
	expectedStrings);
	//============================================
	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0001_m_000002_0 on tt2");
	expectedStrings.put(REDUCE, "attempt_test_0001_r_000002_0 on tt2");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt2",
	expectedStrings);

	taskTrackerManager.killJob(job1.getJobID());
	taskTrackerManager.killJob(job2.getJobID());

	// change configuration
	queues[1].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(25));
	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(75));

	// Re-write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

	refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);

	job1 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[1].getQueueName(), "user");
	job2 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 4, 4,
	queues[2].getQueueName(), "user");

	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0003_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0003_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);


	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0004_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0004_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);


	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0004_m_000002_0 on tt2");
	expectedStrings.put(REDUCE, "attempt_test_0004_r_000002_0 on tt2");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt2",
	expectedStrings);

	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0004_m_000003_0 on tt2");
	expectedStrings.put(REDUCE, "attempt_test_0004_r_000003_0 on tt2");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt2",
	expectedStrings);

	}

	/**
	* Test to verify that the refresh of the scheduler fails when modified
	* configuration overflows 100%
	*
	* @throws Throwable
	*/
	@Test
	public void testFailingCapacityRefresh()
	throws Throwable {

	JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

	queues[0].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(100));
	queues[1].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(70));
	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

	// write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

	try {
	setupAndStartSchedulerFramework(2, 2, 2);
	fail("Scheduler should have failed to start!");
	} catch (IOException ioe) {
	assertTrue(ioe.getMessage().contains(
	String.format(QueueHierarchyBuilder.TOTAL_CAPACITY_OVERFLOWN_MSG,
	queues[1].getQueueName() + "," + queues[2].getQueueName(),
	Float.valueOf(120.0f))));
	}

	// Rectify the properties and start the scheduler
	queues[1].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

	// write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

	setupAndStartSchedulerFramework(2, 2, 2);

	// Now change configuration.
	queues[1].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(35));
	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(95));

	// Re-write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[] { queues[0] });

	try {
	refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);
	} catch (IOException ioe) {
	assertTrue(ioe.getMessage().contains(
	String.format(QueueHierarchyBuilder.TOTAL_CAPACITY_OVERFLOWN_MSG,
	queues[1].getQueueName() + "," + queues[2].getQueueName(),
	Float.valueOf(130.0f))));
	}
	}

	/**
	* @throws Throwable
	*/
	@Test
	public void testRefreshUserLimits()
	throws Throwable {

	JobQueueInfo[] queues = TestQueueManagerRefresh.getSimpleQueueHierarchy();

	queues[0].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(100));
	queues[1].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));
	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.CAPACITY_PROPERTY, String.valueOf(50));

	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
	String.valueOf(100));

	// write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

	setupAndStartSchedulerFramework(1, 2, 2);

	FakeJobInProgress job1 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[2].getQueueName(), "user1");
	FakeJobInProgress job2 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[2].getQueueName(), "user2");

	Map<String, String> expectedStrings = new HashMap<String, String>();
	expectedStrings.put(MAP, "attempt_test_0001_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0001_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);

	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0001_m_000002_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0001_r_000002_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);

	assertNull(scheduler.assignTasks(taskTrackerManager.getTaskTracker("tt1")));
	taskTrackerManager.killJob(job1.getJobID());
	taskTrackerManager.killJob(job2.getJobID());

	// change configuration
	queues[2].getProperties().setProperty(
	CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
	String.valueOf(50));

	// Re-write the configuration file
	QueueManagerTestUtils.writeQueueConfigurationFile(
	queueConfigFile.getAbsolutePath(), new JobQueueInfo[]{queues[0]});

	refreshQueues(taskTrackerManager.getQueueManager(), null, scheduler);

	job1 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[1].getQueueName(), "user1");
	job2 =
	taskTrackerManager.submitJobAndInit(
	JobStatus.PREP, 2, 2,
	queues[2].getQueueName(), "user2");

	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0003_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0003_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);

	expectedStrings.clear();
	expectedStrings.put(MAP, "attempt_test_0004_m_000001_0 on tt1");
	expectedStrings.put(REDUCE, "attempt_test_0004_r_000001_0 on tt1");
	checkMultipleTaskAssignment(
	taskTrackerManager, scheduler, "tt1",
	expectedStrings);
	}

	/**
	* Get a map of all {@link AbstractQueue}s.
	*
	* @param sched
	* @return
	*/
	private static Map<String, AbstractQueue> getAllQueues(
	CapacityTaskScheduler sched) {
	AbstractQueue rootQueue = sched.getRoot();
	HashMap<String, AbstractQueue> allQueues =
	new HashMap<String, AbstractQueue>();
	List<AbstractQueue> allQueuesList = new ArrayList<AbstractQueue>();
	allQueuesList.addAll(rootQueue.getDescendentJobQueues());
	allQueuesList.addAll(rootQueue.getDescendantContainerQueues());
	for (AbstractQueue q : allQueuesList) {
	LOG.info("Putting in allQueues list " + q.getName());
	allQueues.put(q.getName(), q);
	}
	return allQueues;
	}
	}