hadoop-mapreduce/src/test/mapred/org/apache/hadoop/mapred/TestTrackerReservation.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 java.io.IOException;
 import java.util.ArrayList;

 import javax.security.auth.login.LoginException;

 import org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobInProgress;
 import org.apache.hadoop.mapreduce.ClusterMetrics;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;
 import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker;

 import junit.extensions.TestSetup;
 import junit.framework.Test;
 import junit.framework.TestCase;
 import junit.framework.TestSuite;

 public class TestTrackerReservation extends TestCase {

   static String[] trackers = new String[] { "tracker_tracker1:1000",
       "tracker_tracker2:1000", "tracker_tracker3:1000" };
   private static FakeJobTracker jobTracker;

   private static class FakeJobTracker extends
       org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobTracker {

     FakeJobTracker(JobConf conf, Clock clock, String[] tts) throws IOException,
         InterruptedException, LoginException {
       super(conf, clock, tts);
     }

     @Override
     synchronized void finalizeJob(JobInProgress job) {
       // Do nothing
     }
   }


   public static Test suite() {
     TestSetup setup = new TestSetup(new TestSuite(TestTrackerReservation.class)) {
       protected void setUp() throws Exception {
         JobConf conf = new JobConf();
         conf.set(JTConfig.JT_IPC_ADDRESS, "localhost:0");
         conf.set(JTConfig.JT_HTTP_ADDRESS, "0.0.0.0:0");
         jobTracker = new FakeJobTracker(conf, new Clock(), trackers);
         for (String tracker : trackers) {
           FakeObjectUtilities.establishFirstContact(jobTracker, tracker);
         }
       }

       protected void tearDown() throws Exception {
       }
     };
     return setup;
   }

   /**
    * Test case to test if task tracker reservation.
    * <ol>
    * <li>Run a cluster with 3 trackers.</li>
    * <li>Submit a job which reserves all the slots in two
    * trackers.</li>
    * <li>Run the job on another tracker which has
    * no reservations</li>
    * <li>Finish the job and observe the reservations are
    * successfully canceled</li>
    * </ol>
    *
    * @throws Exception
    */
   public void testTaskTrackerReservation() throws Exception {
     JobConf conf = new JobConf();

     conf.setNumMapTasks(1);
     conf.setNumReduceTasks(1);
     conf.setSpeculativeExecution(false);

     conf.setBoolean(JobContext.SETUP_CLEANUP_NEEDED, false);

     //Set task tracker objects for reservation.
     TaskTracker tt1 = jobTracker.getTaskTracker(trackers[0]);
     TaskTracker tt2 = jobTracker.getTaskTracker(trackers[1]);
     TaskTracker tt3 = jobTracker.getTaskTracker(trackers[2]);
     TaskTrackerStatus status1 = new TaskTrackerStatus(
         trackers[0],JobInProgress.convertTrackerNameToHostName(
             trackers[0]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
     TaskTrackerStatus status2 = new TaskTrackerStatus(
         trackers[1],JobInProgress.convertTrackerNameToHostName(
             trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
     TaskTrackerStatus status3 = new TaskTrackerStatus(
         trackers[1],JobInProgress.convertTrackerNameToHostName(
             trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
     tt1.setStatus(status1);
     tt2.setStatus(status2);
     tt3.setStatus(status3);

     FakeJobInProgress fjob = new FakeJobInProgress(conf, jobTracker);
     fjob.setClusterSize(3);
     fjob.initTasks();

     tt1.reserveSlots(TaskType.MAP, fjob, 2);
     tt1.reserveSlots(TaskType.REDUCE, fjob, 2);
     tt3.reserveSlots(TaskType.MAP, fjob, 2);
     tt3.reserveSlots(TaskType.REDUCE, fjob, 2);

     assertEquals("Trackers not reserved for the job : maps",
         2, fjob.getNumReservedTaskTrackersForMaps());
     assertEquals("Trackers not reserved for the job : reduces",
         2, fjob.getNumReservedTaskTrackersForReduces());
     ClusterMetrics metrics = jobTracker.getClusterMetrics();
     assertEquals("reserved map slots do not match",
           4, metrics.getReservedMapSlots());
     assertEquals("reserved reduce slots do not match",
           4, metrics.getReservedReduceSlots());

     TaskAttemptID mTid = fjob.findMapTask(trackers[1]);
     TaskAttemptID rTid = fjob.findReduceTask(trackers[1]);

     fjob.finishTask(mTid);
     fjob.finishTask(rTid);

     assertEquals("Job didnt complete successfully complete", fjob.getStatus()
         .getRunState(), JobStatus.SUCCEEDED);

     assertEquals("Reservation for the job not released: Maps",
         0, fjob.getNumReservedTaskTrackersForMaps());
     assertEquals("Reservation for the job not released : Reduces",
         0, fjob.getNumReservedTaskTrackersForReduces());
     metrics = jobTracker.getClusterMetrics();
     assertEquals("reserved map slots do not match",
         0, metrics.getReservedMapSlots());
     assertEquals("reserved reduce slots do not match",
         0, metrics.getReservedReduceSlots());
   }

   /**
    * Test case to check task tracker reservation for a job which
    * has a job blacklisted tracker.
    * <ol>
    * <li>Run a job which fails on one of the tracker.</li>
    * <li>Check if the job succeeds and has no reservation.</li>
    * </ol>
    *
    * @throws Exception
    */

   public void testTrackerReservationWithJobBlackListedTracker() throws Exception {
     FakeJobInProgress job = TestTaskTrackerBlacklisting.runBlackListingJob(
         jobTracker, trackers);
     assertEquals("Job has no blacklisted trackers", 1, job
         .getBlackListedTrackers().size());
     assertTrue("Tracker 1 not blacklisted for the job", job
         .getBlackListedTrackers().contains(
             JobInProgress.convertTrackerNameToHostName(trackers[0])));
     assertEquals("Job didnt complete successfully complete", job.getStatus()
         .getRunState(), JobStatus.SUCCEEDED);
     assertEquals("Reservation for the job not released: Maps",
         0, job.getNumReservedTaskTrackersForMaps());
     assertEquals("Reservation for the job not released : Reduces",
         0, job.getNumReservedTaskTrackersForReduces());
     ClusterMetrics metrics = jobTracker.getClusterMetrics();
     assertEquals("reserved map slots do not match",
         0, metrics.getReservedMapSlots());
     assertEquals("reserved reduce slots do not match",
         0, metrics.getReservedReduceSlots());
   }

   /**
    * Test case to check if the job reservation is handled properly if the
    * job has a reservation on a black listed tracker.
    *
    * @throws Exception
    */
   public void testReservationOnBlacklistedTracker() throws Exception {
     TaskAttemptID[] taskAttemptID = new TaskAttemptID[3];
     JobConf conf = new JobConf();
     conf.setSpeculativeExecution(false);
     conf.setNumMapTasks(2);
     conf.setNumReduceTasks(2);
     conf.set(JobContext.REDUCE_FAILURES_MAXPERCENT, ".70");
     conf.set(JobContext.MAP_FAILURES_MAX_PERCENT, ".70");
     conf.setBoolean(JobContext.SETUP_CLEANUP_NEEDED, false);
     conf.setMaxTaskFailuresPerTracker(1);
     FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
     job.setClusterSize(trackers.length);
     job.initTasks();

     TaskTracker tt1 = jobTracker.getTaskTracker(trackers[0]);
     TaskTracker tt2 = jobTracker.getTaskTracker(trackers[1]);
     TaskTracker tt3 = jobTracker.getTaskTracker(trackers[2]);
     TaskTrackerStatus status1 = new TaskTrackerStatus(
         trackers[0],JobInProgress.convertTrackerNameToHostName(
             trackers[0]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
     TaskTrackerStatus status2 = new TaskTrackerStatus(
         trackers[1],JobInProgress.convertTrackerNameToHostName(
             trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
     TaskTrackerStatus status3 = new TaskTrackerStatus(
         trackers[1],JobInProgress.convertTrackerNameToHostName(
             trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
     tt1.setStatus(status1);
     tt2.setStatus(status2);
     tt3.setStatus(status3);

     tt1.reserveSlots(TaskType.MAP, job, 2);
     tt1.reserveSlots(TaskType.REDUCE, job, 2);
     tt3.reserveSlots(TaskType.MAP, job, 2);
     tt3.reserveSlots(TaskType.REDUCE, job, 2);

     assertEquals("Trackers not reserved for the job : maps",
         2, job.getNumReservedTaskTrackersForMaps());
     assertEquals("Trackers not reserved for the job : reduces",
         2, job.getNumReservedTaskTrackersForReduces());
     ClusterMetrics metrics = jobTracker.getClusterMetrics();
     assertEquals("reserved map slots do not match",
         4, metrics.getReservedMapSlots());
     assertEquals("reserved reduce slots do not match",
         4, metrics.getReservedReduceSlots());

     /*
      * FakeJobInProgress.findMapTask does not handle
      * task failures. So working around it by failing
      * reduce and blacklisting tracker.
      * Then finish the map task later.
      */
     TaskAttemptID mTid = job.findMapTask(trackers[0]);
     TaskAttemptID rTid = job.findReduceTask(trackers[0]);
     //Task should blacklist the tasktracker.
     job.failTask(rTid);

     assertEquals("Tracker 0 not blacklisted for the job", 1,
         job.getBlackListedTrackers().size());
     assertEquals("Extra Trackers reserved for the job : maps",
         1, job.getNumReservedTaskTrackersForMaps());
     assertEquals("Extra Trackers reserved for the job : reduces",
         1, job.getNumReservedTaskTrackersForReduces());
     metrics = jobTracker.getClusterMetrics();
     assertEquals("reserved map slots do not match",
         2, metrics.getReservedMapSlots());
     assertEquals("reserved reduce slots do not match",
         2, metrics.getReservedReduceSlots());

     //Finish the map task on the tracker 1. Finishing it here to work
     //around bug in the FakeJobInProgress object
     job.finishTask(mTid);
     mTid = job.findMapTask(trackers[1]);
     rTid = job.findReduceTask(trackers[1]);
     job.finishTask(mTid);
     job.finishTask(rTid);
     rTid = job.findReduceTask(trackers[1]);
     job.finishTask(rTid);
     assertEquals("Job didnt complete successfully complete", job.getStatus()
         .getRunState(), JobStatus.SUCCEEDED);
     assertEquals("Trackers not unreserved for the job : maps",
         0, job.getNumReservedTaskTrackersForMaps());
     assertEquals("Trackers not unreserved for the job : reduces",
         0, job.getNumReservedTaskTrackersForReduces());
     metrics = jobTracker.getClusterMetrics();
     assertEquals("reserved map slots do not match",
         0, metrics.getReservedMapSlots());
     assertEquals("reserved reduce slots do not match",
         0, metrics.getReservedReduceSlots());
   }
 }
	/**
	* 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.IOException;
	import java.util.ArrayList;

	import javax.security.auth.login.LoginException;

	import org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobInProgress;
	import org.apache.hadoop.mapreduce.ClusterMetrics;
	import org.apache.hadoop.mapreduce.TaskType;
	import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;
	import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker;

	import junit.extensions.TestSetup;
	import junit.framework.Test;
	import junit.framework.TestCase;
	import junit.framework.TestSuite;

	public class TestTrackerReservation extends TestCase {

	static String[] trackers = new String[] { "tracker_tracker1:1000",
	"tracker_tracker2:1000", "tracker_tracker3:1000" };
	private static FakeJobTracker jobTracker;

	private static class FakeJobTracker extends
	org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobTracker {

	FakeJobTracker(JobConf conf, Clock clock, String[] tts) throws IOException,
	InterruptedException, LoginException {
	super(conf, clock, tts);
	}

	@Override
	synchronized void finalizeJob(JobInProgress job) {
	// Do nothing
	}
	}


	public static Test suite() {
	TestSetup setup = new TestSetup(new TestSuite(TestTrackerReservation.class)) {
	protected void setUp() throws Exception {
	JobConf conf = new JobConf();
	conf.set(JTConfig.JT_IPC_ADDRESS, "localhost:0");
	conf.set(JTConfig.JT_HTTP_ADDRESS, "0.0.0.0:0");
	jobTracker = new FakeJobTracker(conf, new Clock(), trackers);
	for (String tracker : trackers) {
	FakeObjectUtilities.establishFirstContact(jobTracker, tracker);
	}
	}

	protected void tearDown() throws Exception {
	}
	};
	return setup;
	}

	/**
	* Test case to test if task tracker reservation.
	* <ol>
	* <li>Run a cluster with 3 trackers.</li>
	* <li>Submit a job which reserves all the slots in two
	* trackers.</li>
	* <li>Run the job on another tracker which has
	* no reservations</li>
	* <li>Finish the job and observe the reservations are
	* successfully canceled</li>
	* </ol>
	*
	* @throws Exception
	*/
	public void testTaskTrackerReservation() throws Exception {
	JobConf conf = new JobConf();

	conf.setNumMapTasks(1);
	conf.setNumReduceTasks(1);
	conf.setSpeculativeExecution(false);

	conf.setBoolean(JobContext.SETUP_CLEANUP_NEEDED, false);

	//Set task tracker objects for reservation.
	TaskTracker tt1 = jobTracker.getTaskTracker(trackers[0]);
	TaskTracker tt2 = jobTracker.getTaskTracker(trackers[1]);
	TaskTracker tt3 = jobTracker.getTaskTracker(trackers[2]);
	TaskTrackerStatus status1 = new TaskTrackerStatus(
	trackers[0],JobInProgress.convertTrackerNameToHostName(
	trackers[0]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
	TaskTrackerStatus status2 = new TaskTrackerStatus(
	trackers[1],JobInProgress.convertTrackerNameToHostName(
	trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
	TaskTrackerStatus status3 = new TaskTrackerStatus(
	trackers[1],JobInProgress.convertTrackerNameToHostName(
	trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
	tt1.setStatus(status1);
	tt2.setStatus(status2);
	tt3.setStatus(status3);

	FakeJobInProgress fjob = new FakeJobInProgress(conf, jobTracker);
	fjob.setClusterSize(3);
	fjob.initTasks();

	tt1.reserveSlots(TaskType.MAP, fjob, 2);
	tt1.reserveSlots(TaskType.REDUCE, fjob, 2);
	tt3.reserveSlots(TaskType.MAP, fjob, 2);
	tt3.reserveSlots(TaskType.REDUCE, fjob, 2);

	assertEquals("Trackers not reserved for the job : maps",
	2, fjob.getNumReservedTaskTrackersForMaps());
	assertEquals("Trackers not reserved for the job : reduces",
	2, fjob.getNumReservedTaskTrackersForReduces());
	ClusterMetrics metrics = jobTracker.getClusterMetrics();
	assertEquals("reserved map slots do not match",
	4, metrics.getReservedMapSlots());
	assertEquals("reserved reduce slots do not match",
	4, metrics.getReservedReduceSlots());

	TaskAttemptID mTid = fjob.findMapTask(trackers[1]);
	TaskAttemptID rTid = fjob.findReduceTask(trackers[1]);

	fjob.finishTask(mTid);
	fjob.finishTask(rTid);

	assertEquals("Job didnt complete successfully complete", fjob.getStatus()
	.getRunState(), JobStatus.SUCCEEDED);

	assertEquals("Reservation for the job not released: Maps",
	0, fjob.getNumReservedTaskTrackersForMaps());
	assertEquals("Reservation for the job not released : Reduces",
	0, fjob.getNumReservedTaskTrackersForReduces());
	metrics = jobTracker.getClusterMetrics();
	assertEquals("reserved map slots do not match",
	0, metrics.getReservedMapSlots());
	assertEquals("reserved reduce slots do not match",
	0, metrics.getReservedReduceSlots());
	}

	/**
	* Test case to check task tracker reservation for a job which
	* has a job blacklisted tracker.
	* <ol>
	* <li>Run a job which fails on one of the tracker.</li>
	* <li>Check if the job succeeds and has no reservation.</li>
	* </ol>
	*
	* @throws Exception
	*/

	public void testTrackerReservationWithJobBlackListedTracker() throws Exception {
	FakeJobInProgress job = TestTaskTrackerBlacklisting.runBlackListingJob(
	jobTracker, trackers);
	assertEquals("Job has no blacklisted trackers", 1, job
	.getBlackListedTrackers().size());
	assertTrue("Tracker 1 not blacklisted for the job", job
	.getBlackListedTrackers().contains(
	JobInProgress.convertTrackerNameToHostName(trackers[0])));
	assertEquals("Job didnt complete successfully complete", job.getStatus()
	.getRunState(), JobStatus.SUCCEEDED);
	assertEquals("Reservation for the job not released: Maps",
	0, job.getNumReservedTaskTrackersForMaps());
	assertEquals("Reservation for the job not released : Reduces",
	0, job.getNumReservedTaskTrackersForReduces());
	ClusterMetrics metrics = jobTracker.getClusterMetrics();
	assertEquals("reserved map slots do not match",
	0, metrics.getReservedMapSlots());
	assertEquals("reserved reduce slots do not match",
	0, metrics.getReservedReduceSlots());
	}

	/**
	* Test case to check if the job reservation is handled properly if the
	* job has a reservation on a black listed tracker.
	*
	* @throws Exception
	*/
	public void testReservationOnBlacklistedTracker() throws Exception {
	TaskAttemptID[] taskAttemptID = new TaskAttemptID[3];
	JobConf conf = new JobConf();
	conf.setSpeculativeExecution(false);
	conf.setNumMapTasks(2);
	conf.setNumReduceTasks(2);
	conf.set(JobContext.REDUCE_FAILURES_MAXPERCENT, ".70");
	conf.set(JobContext.MAP_FAILURES_MAX_PERCENT, ".70");
	conf.setBoolean(JobContext.SETUP_CLEANUP_NEEDED, false);
	conf.setMaxTaskFailuresPerTracker(1);
	FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
	job.setClusterSize(trackers.length);
	job.initTasks();

	TaskTracker tt1 = jobTracker.getTaskTracker(trackers[0]);
	TaskTracker tt2 = jobTracker.getTaskTracker(trackers[1]);
	TaskTracker tt3 = jobTracker.getTaskTracker(trackers[2]);
	TaskTrackerStatus status1 = new TaskTrackerStatus(
	trackers[0],JobInProgress.convertTrackerNameToHostName(
	trackers[0]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
	TaskTrackerStatus status2 = new TaskTrackerStatus(
	trackers[1],JobInProgress.convertTrackerNameToHostName(
	trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
	TaskTrackerStatus status3 = new TaskTrackerStatus(
	trackers[1],JobInProgress.convertTrackerNameToHostName(
	trackers[1]),0,new ArrayList<TaskStatus>(), 0, 2, 2);
	tt1.setStatus(status1);
	tt2.setStatus(status2);
	tt3.setStatus(status3);

	tt1.reserveSlots(TaskType.MAP, job, 2);
	tt1.reserveSlots(TaskType.REDUCE, job, 2);
	tt3.reserveSlots(TaskType.MAP, job, 2);
	tt3.reserveSlots(TaskType.REDUCE, job, 2);

	assertEquals("Trackers not reserved for the job : maps",
	2, job.getNumReservedTaskTrackersForMaps());
	assertEquals("Trackers not reserved for the job : reduces",
	2, job.getNumReservedTaskTrackersForReduces());
	ClusterMetrics metrics = jobTracker.getClusterMetrics();
	assertEquals("reserved map slots do not match",
	4, metrics.getReservedMapSlots());
	assertEquals("reserved reduce slots do not match",
	4, metrics.getReservedReduceSlots());

	/*
	* FakeJobInProgress.findMapTask does not handle
	* task failures. So working around it by failing
	* reduce and blacklisting tracker.
	* Then finish the map task later.
	*/
	TaskAttemptID mTid = job.findMapTask(trackers[0]);
	TaskAttemptID rTid = job.findReduceTask(trackers[0]);
	//Task should blacklist the tasktracker.
	job.failTask(rTid);

	assertEquals("Tracker 0 not blacklisted for the job", 1,
	job.getBlackListedTrackers().size());
	assertEquals("Extra Trackers reserved for the job : maps",
	1, job.getNumReservedTaskTrackersForMaps());
	assertEquals("Extra Trackers reserved for the job : reduces",
	1, job.getNumReservedTaskTrackersForReduces());
	metrics = jobTracker.getClusterMetrics();
	assertEquals("reserved map slots do not match",
	2, metrics.getReservedMapSlots());
	assertEquals("reserved reduce slots do not match",
	2, metrics.getReservedReduceSlots());

	//Finish the map task on the tracker 1. Finishing it here to work
	//around bug in the FakeJobInProgress object
	job.finishTask(mTid);
	mTid = job.findMapTask(trackers[1]);
	rTid = job.findReduceTask(trackers[1]);
	job.finishTask(mTid);
	job.finishTask(rTid);
	rTid = job.findReduceTask(trackers[1]);
	job.finishTask(rTid);
	assertEquals("Job didnt complete successfully complete", job.getStatus()
	.getRunState(), JobStatus.SUCCEEDED);
	assertEquals("Trackers not unreserved for the job : maps",
	0, job.getNumReservedTaskTrackersForMaps());
	assertEquals("Trackers not unreserved for the job : reduces",
	0, job.getNumReservedTaskTrackersForReduces());
	metrics = jobTracker.getClusterMetrics();
	assertEquals("reserved map slots do not match",
	0, metrics.getReservedMapSlots());
	assertEquals("reserved reduce slots do not match",
	0, metrics.getReservedReduceSlots());
	}
	}