Java 类org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobInProgress 实例源码

/**
 * 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());
}

void addNewTaskStatus(FakeJobInProgress job, TaskType taskType,
      boolean useMapSlot, String tracker, List<TaskStatus> reports) 
      throws IOException {
  TaskAttemptID task = null;
  TaskStatus status = null;
  if (taskType == TaskType.MAP) {
    task = job.findMapTask(tracker);
    status = new MapTaskStatus(task, 0.01f, 2,
          TaskStatus.State.RUNNING, "", "", tracker,
          TaskStatus.Phase.MAP, new Counters());
  } else if (taskType == TaskType.TASK_CLEANUP) {
    if (useMapSlot) {
      status = job.maps[0].taskStatuses.get(
        new TaskAttemptID(job.maps[0].getTIPId(), 0));
    } else {
      status = job.reduces[0].taskStatuses.get(
            new TaskAttemptID(job.reduces[0].getTIPId(), 0));
    }
  } else {
    task = job.findReduceTask(tracker);
    status = new ReduceTaskStatus(task, 0.01f, 2,
          TaskStatus.State.RUNNING, "", "", tracker,
          TaskStatus.Phase.REDUCE, new Counters());
  }
  reports.add(status);
}

/**
 * Test that a setup task can be run against a map slot
 * if it is free.
 * @throws IOException
 */
public void testSetupTaskReturnedForFreeMapSlots() throws IOException {
  // create a job with a setup task.
  FakeJobInProgress job = createJob(TaskType.JOB_SETUP);
  jobTracker.jobs.put(job.getJobID(), job);

  // create a status simulating a free tasktracker
  List<TaskStatus> reports = new ArrayList<TaskStatus>();
  TaskTrackerStatus ttStatus 
    = createTaskTrackerStatus(trackers[2], reports);

  // verify that a setup task can be assigned to a map slot.
  List<Task> tasks = jobTracker.getSetupAndCleanupTasks(ttStatus);
  assertEquals(1, tasks.size());
  assertTrue(tasks.get(0).isJobSetupTask());
  assertTrue(tasks.get(0).isMapTask());
  jobTracker.jobs.clear();
}

/**
 * Test to check that map slots are counted when returning
 * a setup task.
 * @throws IOException
 */
public void testMapSlotsCountedForSetup() throws IOException {
  // create a job with a setup task.
  FakeJobInProgress job = createJob(TaskType.JOB_SETUP);
  jobTracker.jobs.put(job.getJobID(), job);

  // create another job for reservation
  FakeJobInProgress job1 = createJob(null);
  jobTracker.jobs.put(job1.getJobID(), job1);

  // create TT status for testing getSetupAndCleanupTasks
  List<TaskStatus> taskStatuses = new ArrayList<TaskStatus>();
  addNewTaskStatus(job, TaskType.MAP, true, trackers[0], taskStatuses);
  TaskTrackerStatus ttStatus 
    = createTaskTrackerStatus(trackers[0], taskStatuses);

  // test that there should be no map setup task returned.
  List<Task> tasks = jobTracker.getSetupAndCleanupTasks(ttStatus);
  assertEquals(1, tasks.size());
  assertTrue(tasks.get(0).isJobSetupTask());
  assertFalse(tasks.get(0).isMapTask());
  jobTracker.jobs.clear();
}

/**
 * Test to check that reduce slots are also counted when returning
 * a setup task.
 * @throws IOException
 */
public void testReduceSlotsCountedForSetup() throws IOException {
  // create a job with a setup task.
  FakeJobInProgress job = createJob(TaskType.JOB_SETUP);
  jobTracker.jobs.put(job.getJobID(), job);

  // create another job for reservation
  FakeJobInProgress job1 = createJob(null);
  jobTracker.jobs.put(job1.getJobID(), job1);

  // create TT status for testing getSetupAndCleanupTasks
  List<TaskStatus> reports = new ArrayList<TaskStatus>();
  // because free map slots are checked first in code,
  // we fill up map slots also.
  addNewTaskStatus(job1, TaskType.MAP, true, trackers[1], reports);
  addNewTaskStatus(job1, TaskType.REDUCE, false,trackers[1], reports);
  TaskTrackerStatus ttStatus 
    = createTaskTrackerStatus(trackers[1], reports);

  // test that there should be no setup task returned,
  // as both map and reduce slots are occupied.
  List<Task> tasks = jobTracker.getSetupAndCleanupTasks(ttStatus);
  assertNull(tasks);
  jobTracker.jobs.clear();
}

/**
 * Test to check that map slots are counted when returning
 * a taskCleanup task.
 * @throws IOException
 */
public void testNumSlotsUsedForTaskCleanup() throws IOException {
  // Create a high RAM job with a map task's cleanup task and a reduce task's
  // cleanup task. Make this Fake job a high RAM job by setting the slots
  // required for map/reduce task to 2.
  FakeJobInProgress job = createJob(TaskType.TASK_CLEANUP);
  jobTracker.jobs.put(job.getJobID(), job);

  // create TT status for testing getSetupAndCleanupTasks
  List<TaskStatus> taskStatuses = new ArrayList<TaskStatus>();
  TaskTrackerStatus ttStatus =
    createTaskTrackerStatus(trackers[0], taskStatuses);//create dummy status

  // validate mapTaskCleanup task
  validateNumSlotsUsedForTaskCleanup(ttStatus);

  // validate reduceTaskCleanup task
  validateNumSlotsUsedForTaskCleanup(ttStatus);

  jobTracker.jobs.clear();
}

public void testTaskToSpeculate() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(5);
  conf.setNumReduceTasks(5);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWNODE_THRESHOLD, 100f);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  //schedule maps
  taskAttemptID[0] = job.findReduceTask(trackers[0]);
  taskAttemptID[1] = job.findReduceTask(trackers[1]);
  taskAttemptID[2] = job.findReduceTask(trackers[2]);
  taskAttemptID[3] = job.findReduceTask(trackers[3]);
  taskAttemptID[4] = job.findReduceTask(trackers[3]);
  clock.advance(5000);
  job.finishTask(taskAttemptID[0]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[1]);
  clock.advance(20000);
  clock.advanceBySpeculativeLag();
  job.refresh(clock.getTime());
  //we should get a speculative task now
  taskAttemptID[5] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),2);
  clock.advance(5000);
  job.finishTask(taskAttemptID[5]);

  job.refresh(clock.getTime());
  taskAttemptID[5] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),3);
}

public void testTaskLATEScheduling() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[20];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(5);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  clock.advance(2000);
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);
  clock.advance(250000);
  taskAttemptID[4] = job.findMapTask(trackers[3]);
  clock.advanceBySpeculativeLag();
  //by doing the above clock adjustments, we bring the progress rate of
  //taskID 3 lower than 4. For taskID 3, the rate is 85/317000
  //and for taskID 4, the rate is 20/65000. But when we ask for a spec task
  //now, we should get back taskID 4 (since that is expected to complete
  //later than taskID 3).
  job.refresh(clock.getTime());
  job.progressMade(taskAttemptID[3], 0.85f);
  job.progressMade(taskAttemptID[4], 0.20f);
  taskAttemptID[5] = job.findMapTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),4);
}

public void testFastTaskScheduling() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[2];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(2);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  conf.setMapSpeculativeDuration(300L);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  // a really fast task #1
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  clock.advance(2000);
  job.finishTask(taskAttemptID[0]);

  // task #2 is slow
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  clock.advanceBySpeculativeLag();
  clock.advance(5000);
  // 65 secs have elapsed since task scheduling
  // set progress so that it will complete within
  // 300 seconds
  job.progressMade(taskAttemptID[1], 0.7f);

  // no new map task should be found
  job.refresh(clock.getTime());
  assertEquals(job.findMapTask(trackers[2]), null);
}

private int speculativeCap(int totalTasks, int numEarlyComplete, int slots)
throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[1500];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(totalTasks);
  conf.setNumReduceTasks(0);
  jobTracker.setNumSlots(slots);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  int i;
  for (i = 0; i < totalTasks; i++) {
    taskAttemptID[i] = job.findMapTask(trackers[0]);
  }
  clock.advance(5000);
  for (i = 0; i < numEarlyComplete; i++) {
    job.finishTask(taskAttemptID[i]);
  }

  clock.advanceBySpeculativeLag();

  for (i = numEarlyComplete; i < totalTasks; i++) {
    job.progressMade(taskAttemptID[i], 0.85f);
  }
  clock.advance(50000);
  for (i = 0; i < (totalTasks - numEarlyComplete); i++) {
    job.refresh(clock.getTime());
    taskAttemptID[i] = job.findMapTask(trackers[1]);
    clock.advance(2000);
    if (taskAttemptID[i] != null) {
      //add some good progress constantly for the different
      //task-attempts so that
      //the tasktracker doesn't get into the slow trackers category
      job.progressMade(taskAttemptID[i], 0.99f);
    } else {
      break;
    }
  }
  return i;
}

public void testSlowMapProgressingRate() throws IOException {
  clock.advance(1000);
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(3);
  conf.setNumReduceTasks(0);
  //use processing rate for speculation
  conf.setBoolean("mapreduce.job.speculative.using.processing.rate", true);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  //schedule maps
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);

  clock.advance(1000);
  job.finishTask(taskAttemptID[0]);
  //if consider the progress rate, we should speculate task 1
  //but if consider the processing rate, which is map_input_bytes/time
  //then we should speculate task 2
  job.processingRate(taskAttemptID[1], Task.Counter.MAP_INPUT_BYTES,
      100000000, 0.1f, TaskStatus.Phase.MAP);
  job.processingRate(taskAttemptID[2], Task.Counter.MAP_INPUT_BYTES,
      1000, 0.5f, TaskStatus.Phase.MAP);
  clock.advanceBySpeculativeLag();
  //we should get a speculative task now
  job.refresh(clock.getTime());
  taskAttemptID[3] = job.findMapTask(trackers[0]);

  assertEquals(taskAttemptID[3].getTaskID().getId(),2);
}

public void testTaskToSpeculate() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(5);
  conf.setNumReduceTasks(5);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWNODE_THRESHOLD, 100f);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  //schedule maps
  taskAttemptID[0] = job.findReduceTask(trackers[0]);
  taskAttemptID[1] = job.findReduceTask(trackers[1]);
  taskAttemptID[2] = job.findReduceTask(trackers[2]);
  taskAttemptID[3] = job.findReduceTask(trackers[3]);
  taskAttemptID[4] = job.findReduceTask(trackers[3]);
  clock.advance(5000);
  job.finishTask(taskAttemptID[0]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[1]);
  clock.advance(20000);
  clock.advanceBySpeculativeLag();
  job.refresh(clock.getTime());
  //we should get a speculative task now
  taskAttemptID[5] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),2);
  clock.advance(5000);
  job.finishTask(taskAttemptID[5]);

  job.refresh(clock.getTime());
  taskAttemptID[5] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),3);
}

public void testTaskLATEScheduling() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[20];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(5);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  clock.advance(2000);
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);
  clock.advance(250000);
  taskAttemptID[4] = job.findMapTask(trackers[3]);
  clock.advanceBySpeculativeLag();
  //by doing the above clock adjustments, we bring the progress rate of
  //taskID 3 lower than 4. For taskID 3, the rate is 85/317000
  //and for taskID 4, the rate is 20/65000. But when we ask for a spec task
  //now, we should get back taskID 4 (since that is expected to complete
  //later than taskID 3).
  job.refresh(clock.getTime());
  job.progressMade(taskAttemptID[3], 0.85f);
  job.progressMade(taskAttemptID[4], 0.20f);
  taskAttemptID[5] = job.findMapTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),4);
}

public void testFastTaskScheduling() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[2];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(2);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  conf.setMapSpeculativeDuration(300L);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  // a really fast task #1
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  clock.advance(2000);
  job.finishTask(taskAttemptID[0]);

  // task #2 is slow
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  clock.advanceBySpeculativeLag();
  clock.advance(5000);
  // 65 secs have elapsed since task scheduling
  // set progress so that it will complete within
  // 300 seconds
  job.progressMade(taskAttemptID[1], 0.7f);

  // no new map task should be found
  job.refresh(clock.getTime());
  assertEquals(job.findMapTask(trackers[2]), null);
}

private int speculativeCap(int totalTasks, int numEarlyComplete, int slots)
throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[1500];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(totalTasks);
  conf.setNumReduceTasks(0);
  jobTracker.setNumSlots(slots);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  int i;
  for (i = 0; i < totalTasks; i++) {
    taskAttemptID[i] = job.findMapTask(trackers[0]);
  }
  clock.advance(5000);
  for (i = 0; i < numEarlyComplete; i++) {
    job.finishTask(taskAttemptID[i]);
  }

  clock.advanceBySpeculativeLag();

  for (i = numEarlyComplete; i < totalTasks; i++) {
    job.progressMade(taskAttemptID[i], 0.85f);
  }
  clock.advance(50000);
  for (i = 0; i < (totalTasks - numEarlyComplete); i++) {
    job.refresh(clock.getTime());
    taskAttemptID[i] = job.findMapTask(trackers[1]);
    clock.advance(2000);
    if (taskAttemptID[i] != null) {
      //add some good progress constantly for the different
      //task-attempts so that
      //the tasktracker doesn't get into the slow trackers category
      job.progressMade(taskAttemptID[i], 0.99f);
    } else {
      break;
    }
  }
  return i;
}

public void testSlowMapProgressingRate() throws IOException {
  clock.advance(1000);
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(3);
  conf.setNumReduceTasks(0);
  //use processing rate for speculation
  conf.setBoolean("mapreduce.job.speculative.using.processing.rate", true);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  //schedule maps
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);

  clock.advance(1000);
  job.finishTask(taskAttemptID[0]);
  //if consider the progress rate, we should speculate task 1
  //but if consider the processing rate, which is map_input_bytes/time
  //then we should speculate task 2
  job.processingRate(taskAttemptID[1], Task.Counter.MAP_INPUT_BYTES,
      100000000, 0.1f, TaskStatus.Phase.MAP);
  job.processingRate(taskAttemptID[2], Task.Counter.MAP_INPUT_BYTES,
      1000, 0.5f, TaskStatus.Phase.MAP);
  clock.advanceBySpeculativeLag();
  //we should get a speculative task now
  job.refresh(clock.getTime());
  taskAttemptID[3] = job.findMapTask(trackers[0]);

  assertEquals(taskAttemptID[3].getTaskID().getId(),2);
}

public void testLostTracker() throws IOException {
  // Tracker 0 contacts JT
  FakeObjectUtilities.establishFirstContact(jobTracker, trackers[0]);

  TaskAttemptID[] tid = new TaskAttemptID[2];
  JobConf conf = new JobConf();
  conf.setNumMapTasks(1);
  conf.setNumReduceTasks(1);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  // Tracker 0 gets the map task
  tid[0] = job.findMapTask(trackers[0]);

  job.finishTask(tid[0]);

  // Advance clock. Tracker 0 would have got lost
  clock.advance(8 * 1000);

  jobTracker.checkExpiredTrackers();

  // Tracker 1 establishes contact with JT 
  FakeObjectUtilities.establishFirstContact(jobTracker, trackers[1]);

  // Tracker1 should get assigned the lost map task
  tid[1] =  job.findMapTask(trackers[1]);

  assertNotNull("Map Task from Lost Tracker did not get reassigned", tid[1]);

  assertEquals("Task ID of reassigned map task does not match",
      tid[0].getTaskID().toString(), tid[1].getTaskID().toString());

  job.finishTask(tid[1]);

}

public void testTaskToSpeculate() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(5);
  conf.setNumReduceTasks(5);
  conf.setFloat(JobContext.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);    
  job.initTasks();
  //schedule maps
  taskAttemptID[0] = job.findReduceTask(trackers[0]);
  taskAttemptID[1] = job.findReduceTask(trackers[1]);
  taskAttemptID[2] = job.findReduceTask(trackers[2]);
  taskAttemptID[3] = job.findReduceTask(trackers[3]);
  taskAttemptID[4] = job.findReduceTask(trackers[3]);
  clock.advance(5000);
  job.finishTask(taskAttemptID[0]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[1]);
  clock.advance(20000);
  clock.advanceBySpeculativeLag();
  //we should get a speculative task now
  taskAttemptID[5] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),2);
  clock.advance(5000);
  job.finishTask(taskAttemptID[5]);

  taskAttemptID[5] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),3);

  // Verify total speculative tasks by jobtracker instrumentation
  assertEquals("Total speculative maps", 1, fakeInst.numSpeculativeMaps);
  assertEquals("Total speculative reduces", 3,
               fakeInst.numSpeculativeReduces);
  LOG.info("Total speculative maps = " + fakeInst.numSpeculativeMaps);
  LOG.info("Total speculative reduces = " + fakeInst.numSpeculativeReduces);
}

public void testTaskLATEScheduling() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[20];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(5);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobContext.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  clock.advance(2000);
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);
  clock.advance(250000);
  taskAttemptID[4] = job.findMapTask(trackers[3]);
  clock.advanceBySpeculativeLag();
  //by doing the above clock adjustments, we bring the progress rate of 
  //taskID 3 lower than 4. For taskID 3, the rate is 85/317000
  //and for taskID 4, the rate is 20/65000. But when we ask for a spec task
  //now, we should get back taskID 4 (since that is expected to complete
  //later than taskID 3).
  job.progressMade(taskAttemptID[3], 0.85f);
  job.progressMade(taskAttemptID[4], 0.20f);
  taskAttemptID[5] = job.findMapTask(trackers[4]);
  assertEquals(taskAttemptID[5].getTaskID().getId(),4);
  // Verify total speculative tasks by jobtracker instrumentation
  assertEquals("Total speculative maps", 2, fakeInst.numSpeculativeMaps);
  assertEquals("Total speculative reduces", 3,
               fakeInst.numSpeculativeReduces);
  LOG.info("Total speculative maps = " + fakeInst.numSpeculativeMaps);
  LOG.info("Total speculative reduces = " + fakeInst.numSpeculativeReduces);
}

private int speculativeCap(int totalTasks, int numEarlyComplete, int slots)
throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[1500];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(totalTasks);
  conf.setNumReduceTasks(0);
  jobTracker.setNumSlots(slots);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  int i;
  for (i = 0; i < totalTasks; i++) {
    taskAttemptID[i] = job.findMapTask(trackers[0]);
  }
  clock.advance(5000);
  for (i = 0; i < numEarlyComplete; i++) {
    job.finishTask(taskAttemptID[i]);
  }

  clock.advanceBySpeculativeLag();

  for (i = numEarlyComplete; i < totalTasks; i++) {
    job.progressMade(taskAttemptID[i], 0.85f);
  }
  clock.advance(50000);
  for (i = 0; i < (totalTasks - numEarlyComplete); i++) {
    taskAttemptID[i] = job.findMapTask(trackers[1]);
    clock.advance(2000);
    if (taskAttemptID[i] != null) {
      //add some good progress constantly for the different
      //task-attempts so that
      //the tasktracker doesn't get into the slow trackers category
      job.progressMade(taskAttemptID[i], 0.99f);
    } else {
      break;
    }
  }
  return i;
}

public void testRunningTaskCountWithSpeculation() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[8];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(3);
  conf.setNumReduceTasks(3);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  //Check for runningMap counts first
  //schedule maps
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);

  clock.advance(5000);
  job.finishTask(taskAttemptID[0]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[1]);
  clock.advanceBySpeculativeLag();

  //we should get a speculative task now
  job.refresh(clock.getTime());
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  job.refresh(clock.getTime());
  int oldRunningMap = job.runningMaps();
  LOG.info("No of running maps before fail was " + oldRunningMap);
  job.failTask(taskAttemptID[2]);
  job.refresh(clock.getTime());
  assertEquals(
    "Running maps count should be updated from " + oldRunningMap + " to " +
      (oldRunningMap - 1), job.runningMaps(), oldRunningMap - 1);
  LOG.info(" Job running maps after fail " + job.runningMaps());

  clock.advance(5000);
  job.finishTask(taskAttemptID[3]);

  //check for runningReduce count.
  taskAttemptID[4] = job.findReduceTask(trackers[0]);
  taskAttemptID[5] = job.findReduceTask(trackers[1]);
  taskAttemptID[6] = job.findReduceTask(trackers[2]);

  clock.advance(5000);
  job.finishTask(taskAttemptID[4]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[5]);

  job.refresh(clock.getTime());
  clock.advanceBySpeculativeLag();
  taskAttemptID[7] = job.findReduceTask(trackers[4]);

  job.refresh(clock.getTime());
  int oldRunningReduces = job.runningReduces();
  job.failTask(taskAttemptID[6]);
  job.refresh(clock.getTime());
  LOG.info(
    " No of running Reduces before fail " + oldRunningReduces);
  LOG.info(
    " No of runing reduces after fail " + job.runningReduces());
  assertEquals(
    "Running reduces count should be updated from " + oldRunningReduces +
      " to " + (oldRunningReduces - 1), job.runningReduces(),
    oldRunningReduces - 1);

  job.finishTask(taskAttemptID[7]);
}

public void testIsSlowTracker() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[20];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(10);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  //schedule some tasks
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[0]);
  taskAttemptID[2] = job.findMapTask(trackers[0]);
  taskAttemptID[3] = job.findMapTask(trackers[1]);
  taskAttemptID[4] = job.findMapTask(trackers[1]);
  taskAttemptID[5] = job.findMapTask(trackers[1]);
  taskAttemptID[6] = job.findMapTask(trackers[2]);
  taskAttemptID[7] = job.findMapTask(trackers[2]);
  taskAttemptID[8] = job.findMapTask(trackers[2]);
  clock.advance(1000);
  //Some tasks finish in 1 second (on trackers[0])
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);
  clock.advance(1000);
  //Some tasks finish in 2 second (on trackers[1])
  job.finishTask(taskAttemptID[3]);
  job.finishTask(taskAttemptID[4]);
  job.finishTask(taskAttemptID[5]);
  assertEquals("Tracker "+ trackers[0] + " expected to be not slow ",
      job.isSlowTracker(trackers[0]), false);
  clock.advance(100000);
  //After a long time, some tasks finished on trackers[2]
  job.finishTask(taskAttemptID[6]);
  job.finishTask(taskAttemptID[7]);
  job.finishTask(taskAttemptID[8]);
  job.refresh(clock.getTime());
  assertEquals("Tracker "+ trackers[2] + " expected to be slow ",
      job.isSlowTracker(trackers[2]), true);
}

/**
 * tests that a task that has a remaining time less than duration
 * time
 */
public void testTaskSpeculationStddevCap() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[8];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setFloat(JobInProgress.SPECULATIVE_STDDEVMEANRATIO_MAX, 0.33f);
  conf.setNumMapTasks(7);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  // all but one tasks start off
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);
  taskAttemptID[3] = job.findMapTask(trackers[0]);
  taskAttemptID[4] = job.findMapTask(trackers[1]);
  taskAttemptID[5] = job.findMapTask(trackers[2]);

  // 3 tasks finish really fast in 15s
  clock.advance (15 * 1000);
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);

  // advance to 600s and schedule last mapper
  clock.advance (585 * 1000);
  taskAttemptID[6] = job.findMapTask(trackers[0]);

  // advance to 700s and report progress
  clock.advance (10 * 60 * 1000);

  // set progress rates
  job.progressMade(taskAttemptID[3], 0.2f);
  job.progressMade(taskAttemptID[4], 0.5f);
  job.progressMade(taskAttemptID[5], 0.6f);
  job.progressMade(taskAttemptID[6], 0.02f);

  // the progress has been set in such a way that
  // stddev > mean. now we depend on stddev capping
  // for speculation.

  job.refresh(clock.getTime());
  taskAttemptID[7] = job.findMapTask(trackers[1]);

  // no new map task should be found
  if(taskAttemptID[7] ==  null)
    Assert.fail();
}

public void testSpeculateLastTask() throws Exception {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[8];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(3);
  conf.setNumReduceTasks(3);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);

  clock.advanceBySpeculativeLag();
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);

  // Speculative last unfinised task
  job.refresh(clock.getTime());
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  Assert.assertNotNull(taskAttemptID[3]);

  job.finishTask(taskAttemptID[2]);
  job.finishTask(taskAttemptID[3]);

  taskAttemptID[4] = job.findReduceTask(trackers[0]);
  taskAttemptID[5] = job.findReduceTask(trackers[1]);
  taskAttemptID[6] = job.findReduceTask(trackers[2]);

  clock.advanceBySpeculativeLag();
  job.finishTask(taskAttemptID[4]);
  job.finishTask(taskAttemptID[5]);

  // Speculative last unfinised task
  job.refresh(clock.getTime());
  taskAttemptID[7] = job.findReduceTask(trackers[3]);
  Assert.assertNotNull(taskAttemptID[7]);

  job.finishTask(taskAttemptID[6]);
  job.finishTask(taskAttemptID[7]);
}

public void testSlowReduceProgressingRate() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(4);
  conf.setNumReduceTasks(4);
  //use processing rate for speculation
  conf.setBoolean("mapreduce.job.speculative.using.processing.rate", true);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  //schedule reduces
  taskAttemptID[0] = job.findReduceTask(trackers[0]);
  taskAttemptID[1] = job.findReduceTask(trackers[1]);
  taskAttemptID[2] = job.findReduceTask(trackers[2]);
  taskAttemptID[3] = job.findReduceTask(trackers[3]);

  clock.advance(1000);

  //task 0 just starts copying, while task 1, 2, 3 are already in the reducing
  //phase. If we compared the progress rate, then we should speculate 0.
  //However, by comparing the processing rate in the copy phase, among all 4
  //tasks, task 0 is fast, and we should not speculate it.
  //for task 1, 2, 3, they are all in the reducing phase, with same progress,
  //however, task 1 has smaller processing rate(the statistics of the reduce
  //phase for all the tasks will also include statistics for task 0, whose
  //processing rate is 0)
  job.finishCopy(taskAttemptID[1], clock.getTime(), 10000);
  job.finishCopy(taskAttemptID[2], clock.getTime(), 10000);
  job.finishCopy(taskAttemptID[3], clock.getTime(), 10000);
  clock.advance(1000);
  job.finishSort(taskAttemptID[1], clock.getTime());
  job.finishSort(taskAttemptID[2], clock.getTime());
  job.finishSort(taskAttemptID[3], clock.getTime());
  job.processingRate(taskAttemptID[0], Task.Counter.REDUCE_SHUFFLE_BYTES,
      100000000, 0.1f, TaskStatus.Phase.SHUFFLE);
  job.processingRate(taskAttemptID[1], Task.Counter.REDUCE_INPUT_BYTES,
      1000, 0.8f, TaskStatus.Phase.REDUCE);
  job.processingRate(taskAttemptID[2], Task.Counter.REDUCE_INPUT_BYTES,
      100000000, 0.8f, TaskStatus.Phase.REDUCE);
  job.processingRate(taskAttemptID[3], Task.Counter.REDUCE_INPUT_BYTES,
      100000000, 0.8f, TaskStatus.Phase.REDUCE);
  clock.advanceBySpeculativeLag();
  //we should get a speculative task now
  job.refresh(clock.getTime());
  taskAttemptID[4] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[4].getTaskID().getId(),1);
}

public void testRunningTaskCountWithSpeculation() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[8];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(3);
  conf.setNumReduceTasks(3);
  conf.setFloat(JobInProgress.SPECULATIVE_SLOWTASK_THRESHOLD, 0.5f);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  //Check for runningMap counts first
  //schedule maps
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);

  clock.advance(5000);
  job.finishTask(taskAttemptID[0]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[1]);
  clock.advanceBySpeculativeLag();

  //we should get a speculative task now
  job.refresh(clock.getTime());
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  job.refresh(clock.getTime());
  int oldRunningMap = job.runningMaps();
  LOG.info("No of running maps before fail was " + oldRunningMap);
  job.failTask(taskAttemptID[2]);
  job.refresh(clock.getTime());
  assertEquals(
    "Running maps count should be updated from " + oldRunningMap + " to " +
      (oldRunningMap - 1), job.runningMaps(), oldRunningMap - 1);
  LOG.info(" Job running maps after fail " + job.runningMaps());

  clock.advance(5000);
  job.finishTask(taskAttemptID[3]);

  //check for runningReduce count.
  taskAttemptID[4] = job.findReduceTask(trackers[0]);
  taskAttemptID[5] = job.findReduceTask(trackers[1]);
  taskAttemptID[6] = job.findReduceTask(trackers[2]);

  clock.advance(5000);
  job.finishTask(taskAttemptID[4]);
  clock.advance(1000);
  job.finishTask(taskAttemptID[5]);

  job.refresh(clock.getTime());
  clock.advanceBySpeculativeLag();
  taskAttemptID[7] = job.findReduceTask(trackers[4]);

  job.refresh(clock.getTime());
  int oldRunningReduces = job.runningReduces();
  job.failTask(taskAttemptID[6]);
  job.refresh(clock.getTime());
  LOG.info(
    " No of running Reduces before fail " + oldRunningReduces);
  LOG.info(
    " No of runing reduces after fail " + job.runningReduces());
  assertEquals(
    "Running reduces count should be updated from " + oldRunningReduces +
      " to " + (oldRunningReduces - 1), job.runningReduces(),
    oldRunningReduces - 1);

  job.finishTask(taskAttemptID[7]);
}

public void testIsSlowTracker() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[20];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(10);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();
  //schedule some tasks
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[0]);
  taskAttemptID[2] = job.findMapTask(trackers[0]);
  taskAttemptID[3] = job.findMapTask(trackers[1]);
  taskAttemptID[4] = job.findMapTask(trackers[1]);
  taskAttemptID[5] = job.findMapTask(trackers[1]);
  taskAttemptID[6] = job.findMapTask(trackers[2]);
  taskAttemptID[7] = job.findMapTask(trackers[2]);
  taskAttemptID[8] = job.findMapTask(trackers[2]);
  clock.advance(1000);
  //Some tasks finish in 1 second (on trackers[0])
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);
  clock.advance(1000);
  //Some tasks finish in 2 second (on trackers[1])
  job.finishTask(taskAttemptID[3]);
  job.finishTask(taskAttemptID[4]);
  job.finishTask(taskAttemptID[5]);
  assertEquals("Tracker "+ trackers[0] + " expected to be not slow ",
      job.isSlowTracker(trackers[0]), false);
  clock.advance(100000);
  //After a long time, some tasks finished on trackers[2]
  job.finishTask(taskAttemptID[6]);
  job.finishTask(taskAttemptID[7]);
  job.finishTask(taskAttemptID[8]);
  job.refresh(clock.getTime());
  assertEquals("Tracker "+ trackers[2] + " expected to be slow ",
      job.isSlowTracker(trackers[2]), true);
}

/**
 * tests that a task that has a remaining time less than duration
 * time
 */
public void testTaskSpeculationStddevCap() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[8];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setFloat(JobInProgress.SPECULATIVE_STDDEVMEANRATIO_MAX, 0.33f);
  conf.setNumMapTasks(7);
  conf.setNumReduceTasks(0);
  conf.setFloat(JobInProgress.SPECULATIVE_MAP_UNFINISHED_THRESHOLD_KEY, 0);
  conf.setFloat(JobInProgress.SPECULATIVE_REDUCE_UNFINISHED_THRESHOLD_KEY, 0);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  // all but one tasks start off
  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);
  taskAttemptID[3] = job.findMapTask(trackers[0]);
  taskAttemptID[4] = job.findMapTask(trackers[1]);
  taskAttemptID[5] = job.findMapTask(trackers[2]);

  // 3 tasks finish really fast in 15s
  clock.advance (15 * 1000);
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);
  job.finishTask(taskAttemptID[2]);

  // advance to 600s and schedule last mapper
  clock.advance (585 * 1000);
  taskAttemptID[6] = job.findMapTask(trackers[0]);

  // advance to 700s and report progress
  clock.advance (10 * 60 * 1000);

  // set progress rates
  job.progressMade(taskAttemptID[3], 0.2f);
  job.progressMade(taskAttemptID[4], 0.5f);
  job.progressMade(taskAttemptID[5], 0.6f);
  job.progressMade(taskAttemptID[6], 0.02f);

  // the progress has been set in such a way that
  // stddev > mean. now we depend on stddev capping
  // for speculation.

  job.refresh(clock.getTime());
  taskAttemptID[7] = job.findMapTask(trackers[1]);

  // no new map task should be found
  if(taskAttemptID[7] ==  null)
    Assert.fail();
}

public void testSpeculateLastTask() throws Exception {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[8];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(3);
  conf.setNumReduceTasks(3);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  taskAttemptID[0] = job.findMapTask(trackers[0]);
  taskAttemptID[1] = job.findMapTask(trackers[1]);
  taskAttemptID[2] = job.findMapTask(trackers[2]);

  clock.advanceBySpeculativeLag();
  job.finishTask(taskAttemptID[0]);
  job.finishTask(taskAttemptID[1]);

  // Speculative last unfinised task
  job.refresh(clock.getTime());
  taskAttemptID[3] = job.findMapTask(trackers[3]);
  Assert.assertNotNull(taskAttemptID[3]);

  job.finishTask(taskAttemptID[2]);
  job.finishTask(taskAttemptID[3]);

  taskAttemptID[4] = job.findReduceTask(trackers[0]);
  taskAttemptID[5] = job.findReduceTask(trackers[1]);
  taskAttemptID[6] = job.findReduceTask(trackers[2]);

  clock.advanceBySpeculativeLag();
  job.finishTask(taskAttemptID[4]);
  job.finishTask(taskAttemptID[5]);

  // Speculative last unfinised task
  job.refresh(clock.getTime());
  taskAttemptID[7] = job.findReduceTask(trackers[3]);
  Assert.assertNotNull(taskAttemptID[7]);

  job.finishTask(taskAttemptID[6]);
  job.finishTask(taskAttemptID[7]);
}

public void testSlowReduceProgressingRate() throws IOException {
  TaskAttemptID[] taskAttemptID = new TaskAttemptID[6];
  JobConf conf = new JobConf();
  conf.setSpeculativeExecution(true);
  conf.setNumMapTasks(4);
  conf.setNumReduceTasks(4);
  //use processing rate for speculation
  conf.setBoolean("mapreduce.job.speculative.using.processing.rate", true);
  FakeJobInProgress job = new FakeJobInProgress(conf, jobTracker);
  job.initTasks();

  //schedule reduces
  taskAttemptID[0] = job.findReduceTask(trackers[0]);
  taskAttemptID[1] = job.findReduceTask(trackers[1]);
  taskAttemptID[2] = job.findReduceTask(trackers[2]);
  taskAttemptID[3] = job.findReduceTask(trackers[3]);

  clock.advance(1000);

  //task 0 just starts copying, while task 1, 2, 3 are already in the reducing
  //phase. If we compared the progress rate, then we should speculate 0.
  //However, by comparing the processing rate in the copy phase, among all 4
  //tasks, task 0 is fast, and we should not speculate it.
  //for task 1, 2, 3, they are all in the reducing phase, with same progress,
  //however, task 1 has smaller processing rate(the statistics of the reduce
  //phase for all the tasks will also include statistics for task 0, whose
  //processing rate is 0)
  job.finishCopy(taskAttemptID[1], clock.getTime(), 10000);
  job.finishCopy(taskAttemptID[2], clock.getTime(), 10000);
  job.finishCopy(taskAttemptID[3], clock.getTime(), 10000);
  clock.advance(1000);
  job.finishSort(taskAttemptID[1], clock.getTime());
  job.finishSort(taskAttemptID[2], clock.getTime());
  job.finishSort(taskAttemptID[3], clock.getTime());
  job.processingRate(taskAttemptID[0], Task.Counter.REDUCE_SHUFFLE_BYTES,
      100000000, 0.1f, TaskStatus.Phase.SHUFFLE);
  job.processingRate(taskAttemptID[1], Task.Counter.REDUCE_INPUT_BYTES,
      1000, 0.8f, TaskStatus.Phase.REDUCE);
  job.processingRate(taskAttemptID[2], Task.Counter.REDUCE_INPUT_BYTES,
      100000000, 0.8f, TaskStatus.Phase.REDUCE);
  job.processingRate(taskAttemptID[3], Task.Counter.REDUCE_INPUT_BYTES,
      100000000, 0.8f, TaskStatus.Phase.REDUCE);
  clock.advanceBySpeculativeLag();
  //we should get a speculative task now
  job.refresh(clock.getTime());
  taskAttemptID[4] = job.findReduceTask(trackers[4]);
  assertEquals(taskAttemptID[4].getTaskID().getId(),1);
}

/**
 * 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());
}