/**
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.Cluster.JobTrackerStatus;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.QueueState;
import static org.apache.hadoop.mapred.QueueManager.toFullPropertyName;
import org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobHistory;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;
import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker;
import org.apache.hadoop.mapreduce.split.JobSplit;
import org.apache.hadoop.security.authorize.AccessControlList;


public class CapacityTestUtils {
  static final Log LOG =
    LogFactory.getLog(org.apache.hadoop.mapred.CapacityTestUtils.class);
  static final String MAP = "map";
  static final String REDUCE = "reduce";


  /**
   * Test class that removes the asynchronous nature of job initialization.
   * <p/>
   * The run method is a dummy which just waits for completion. It is
   * expected that test code calls the main method, initializeJobs, directly
   * to trigger initialization.
   */
  static class ControlledJobInitializer extends
                              JobInitializationPoller.JobInitializationThread {

    boolean stopRunning;

    public ControlledJobInitializer(JobInitializationPoller p) {
      p.super();
    }

    @Override
    public void run() {
      while (!stopRunning) {
        try {
          synchronized (this) {
            this.wait();
          }
        } catch (InterruptedException ie) {
          break;
        }
      }
    }

    void stopRunning() {
      stopRunning = true;
    }
  }


  /**
   * Test class that removes the asynchronous nature of job initialization.
   * <p/>
   * The run method is a dummy which just waits for completion. It is
   * expected that test code calls the main method, selectJobsToInitialize,
   * directly to trigger initialization.
   * <p/>
   * The class also creates the test worker thread objects of type
   * ControlledJobInitializer instead of the objects of the actual class
   */
  static class ControlledInitializationPoller extends JobInitializationPoller {

    private boolean stopRunning;
    private ArrayList<ControlledJobInitializer> workers;

    public ControlledInitializationPoller(JobQueuesManager mgr,
        TaskTrackerManager ttm) {
      super(mgr, ttm);
    }

    @Override
    public void run() {
      // don't do anything here.
      while (!stopRunning) {
        try {
          synchronized (this) {
            this.wait();
          }
        } catch (InterruptedException ie) {
          break;
        }
      }
    }

    @Override
    JobInitializationThread createJobInitializationThread() {
      ControlledJobInitializer t = new ControlledJobInitializer(this);
      if (workers == null) {
        workers = new ArrayList<ControlledJobInitializer>();
      }
      workers.add(t);
      return t;
    }

    @Override
    void selectJobsToInitialize() {
      super.cleanUpInitializedJobsList();
      super.selectJobsToInitialize();
      for (ControlledJobInitializer t : workers) {
        t.initializeJobs();
      }
    }

    void stopRunning() {
      stopRunning = true;
      for (ControlledJobInitializer t : workers) {
        t.stopRunning();
        t.interrupt();
      }
    }
  }

  static class FakeClock extends CapacityTaskScheduler.Clock {
    private long time = 0;

    public void advance(long millis) {
      time += millis;
    }

    @Override
    long getTime() {
      return time;
    }
  }

  /**
   * The method accepts a attempt string and checks for validity of
   * assignTask w.r.t attempt string.
   * 
   * @param taskTrackerManager
   * @param scheduler
   * @param taskTrackerName
   * @param expectedTaskString
   * @return
   * @throws IOException
   */
  static Task checkAssignment(
    CapacityTestUtils.FakeTaskTrackerManager taskTrackerManager,
    CapacityTaskScheduler scheduler, String taskTrackerName,
    String expectedTaskString) throws IOException {
    Map<String, String> expectedStrings = new HashMap<String, String>();
    if (expectedTaskString.contains("_m_")) {
      expectedStrings.put(MAP, expectedTaskString);
    } else if (expectedTaskString.contains("_r_")) {
      expectedStrings.put(REDUCE, expectedTaskString);
    }
    List<Task> tasks = checkMultipleTaskAssignment(
      taskTrackerManager, scheduler, taskTrackerName, expectedStrings);
    for (Task task : tasks) {
      if (task.toString().equals(expectedTaskString)) {
        return task;
      }
    }
    return null;
  }

  /**
   * Checks the validity of tasks assigned by scheduler's assignTasks method
   * According to JIRA:1030 every assignTasks call in CapacityScheduler
   * would result in either MAP or REDUCE or BOTH.
   *
   * This method accepts a Map<String,String>.
   * The map should always have <=2 entried in hashMap.
   *
   * sample calling code .
   *
   *  Map<String, String> expectedStrings = new HashMap<String, String>();
   * ......
   * .......
   * expectedStrings.clear();
   * 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);
   * 
   * @param taskTrackerManager
   * @param scheduler
   * @param taskTrackerName
   * @param expectedTaskStrings
   * @return
   * @throws IOException
   */
  static List<Task> checkMultipleTaskAssignment(
    CapacityTestUtils.FakeTaskTrackerManager taskTrackerManager,
    CapacityTaskScheduler scheduler, String taskTrackerName,
    Map<String,String> expectedTaskStrings) throws IOException {
    //Call assign task
    List<Task> tasks = scheduler.assignTasks(
      taskTrackerManager.getTaskTracker(
        taskTrackerName));

    if (tasks==null || tasks.isEmpty()) {
      if (expectedTaskStrings.size() > 0) {
        fail("Expected some tasks to be assigned, but got none.");  
      } else {
        return null;
      }
    }

    if (expectedTaskStrings.size() > tasks.size()) {
      StringBuffer sb = new StringBuffer();
      sb.append("Expected strings different from actual strings.");
      sb.append(" Expected string count=").append(expectedTaskStrings.size());
      sb.append(" Actual string count=").append(tasks.size());
      sb.append(" Expected strings=");
      for (String expectedTask : expectedTaskStrings.values()) {
        sb.append(expectedTask).append(",");
      }
      sb.append("Actual strings=");
      for (Task actualTask : tasks) {
        sb.append(actualTask.toString()).append(",");
      }
      fail(sb.toString());
    }
    
    for (Task task : tasks) {
      LOG.info("tasks are : " + tasks.toString());
      if (task.isMapTask()) {
        //check if expected string is set for map or not.
        if (expectedTaskStrings.get(MAP) != null) {
          assertEquals(expectedTaskStrings.get(MAP), task.toString());
        } else {
          fail("No map task is expected, but got " + task.toString());
        }
      } else {
        //check if expectedStrings is set for reduce or not.
        if (expectedTaskStrings.get(REDUCE) != null) {
          assertEquals(expectedTaskStrings.get(REDUCE), task.toString());
        } else {
          fail("No reduce task is expected, but got " + task.toString());
        }
      }
    }
    return tasks;
  }

  static void verifyCapacity(
    FakeTaskTrackerManager taskTrackerManager,
    String expectedCapacity,
    String queue) throws IOException {
    String schedInfo = taskTrackerManager.getQueueManager().
      getSchedulerInfo(queue).toString();
    assertTrue(
      schedInfo.contains(
        "Map tasks\nCapacity: "
          + expectedCapacity + " slots"));
  }

  /*
  * Fake job in progress object used for testing the schedulers scheduling
  * decisions. The JobInProgress objects returns out FakeTaskInProgress
  * objects when assignTasks is called. If speculative maps and reduces
  * are configured then JobInProgress returns exactly one Speculative
  * map and reduce task.
  */
  static class FakeJobInProgress extends JobInProgress {

    protected FakeTaskTrackerManager taskTrackerManager;
    private int mapTaskCtr;
    private int redTaskCtr;
    private Set<TaskInProgress> mapTips =
      new HashSet<TaskInProgress>();
    private Set<TaskInProgress> reduceTips =
      new HashSet<TaskInProgress>();
    private int speculativeMapTaskCounter = 0;
    private int speculativeReduceTaskCounter = 0;

    public FakeJobInProgress(
      JobID jId, JobConf jobConf,
      FakeTaskTrackerManager taskTrackerManager, String user, 
      JobTracker jt) throws IOException {
      super(jId, jobConf, jt);
      this.taskTrackerManager = taskTrackerManager;
      this.startTime = System.currentTimeMillis();
      this.status = new JobStatus(
        jId, 0f, 0f, JobStatus.PREP,
        jobConf.getUser(),
        jobConf.getJobName(), "", "");
      this.status.setJobPriority(JobPriority.NORMAL);
      this.status.setStartTime(startTime);
      if (null == jobConf.getQueueName()) {
        this.profile = new JobProfile(
          user, jId,
          null, null, null);
      } else {
        this.profile = new JobProfile(
          user, jId,
          null, null, null, jobConf.getQueueName());
      }
      mapTaskCtr = 0;
      redTaskCtr = 0;
      this.jobHistory = new FakeJobHistory();
    }

    @Override
    public synchronized void initTasks() throws IOException {
      getStatus().setRunState(JobStatus.RUNNING);
    }

    @Override
    public synchronized Task obtainNewMapTask(
      final TaskTrackerStatus tts, int clusterSize,
      int ignored) throws IOException {
      boolean areAllMapsRunning = (mapTaskCtr == numMapTasks);
      if (areAllMapsRunning) {
        if (!getJobConf().getMapSpeculativeExecution() ||
          speculativeMapTasks > 0) {
          return null;
        }
      }
      TaskAttemptID attemptId = getTaskAttemptID(true, areAllMapsRunning);
      JobSplit.TaskSplitMetaInfo split = JobSplit.EMPTY_TASK_SPLIT;
      Task task = new MapTask(
        "", attemptId, 0, split.getSplitIndex(), super.numSlotsPerMap) {
        @Override
        public String toString() {
          return String.format("%s on %s", getTaskID(), tts.getTrackerName());
        }
      };
      taskTrackerManager.startTask(tts.getTrackerName(), task);
      runningMapTasks++;
      // create a fake TIP and keep track of it
      FakeTaskInProgress mapTip = new FakeTaskInProgress(
        getJobID(),
        getJobConf(), task, true, this, split);
      mapTip.taskStatus.setRunState(TaskStatus.State.RUNNING);
      if (areAllMapsRunning) {
        speculativeMapTasks++;
        //you have scheduled a speculative map. Now set all tips in the
        //map tips not to have speculative task.
        for (TaskInProgress t : mapTips) {
          if (t instanceof FakeTaskInProgress) {
            FakeTaskInProgress mt = (FakeTaskInProgress) t;
            mt.hasSpeculativeMap = false;
          }
        }
      } else {
        //add only non-speculative tips.
        mapTips.add(mapTip);
        //add the tips to the JobInProgress TIPS
        maps = mapTips.toArray(new TaskInProgress[mapTips.size()]);
      }
      return task;
    }

    @Override
    public synchronized Task obtainNewReduceTask(
      final TaskTrackerStatus tts,
      int clusterSize, int ignored) throws IOException {
      boolean areAllReducesRunning = (redTaskCtr == numReduceTasks);
      if (areAllReducesRunning) {
        if (!getJobConf().getReduceSpeculativeExecution() ||
          speculativeReduceTasks > 0) {
          return null;
        }
      }
      TaskAttemptID attemptId = getTaskAttemptID(false, areAllReducesRunning);
      Task task = new ReduceTask(
        "", attemptId, 0, 10,
        super.numSlotsPerReduce) {
        @Override
        public String toString() {
          return String.format("%s on %s", getTaskID(), tts.getTrackerName());
        }
      };
      taskTrackerManager.startTask(tts.getTrackerName(), task);
      runningReduceTasks++;
      // create a fake TIP and keep track of it
      FakeTaskInProgress reduceTip = new FakeTaskInProgress(
        getJobID(),
        getJobConf(), task, false, this, null);
      reduceTip.taskStatus.setRunState(TaskStatus.State.RUNNING);
      if (areAllReducesRunning) {
        speculativeReduceTasks++;
        //you have scheduled a speculative map. Now set all tips in the
        //map tips not to have speculative task.
        for (TaskInProgress t : reduceTips) {
          if (t instanceof FakeTaskInProgress) {
            FakeTaskInProgress rt = (FakeTaskInProgress) t;
            rt.hasSpeculativeReduce = false;
          }
        }
      } else {
        //add only non-speculative tips.
        reduceTips.add(reduceTip);
        //add the tips to the JobInProgress TIPS
        reduces = reduceTips.toArray(new TaskInProgress[reduceTips.size()]);
      }
      return task;
    }

    public void mapTaskFinished() {
      runningMapTasks--;
      finishedMapTasks++;
    }

    public void reduceTaskFinished() {
      runningReduceTasks--;
      finishedReduceTasks++;
    }

    private TaskAttemptID getTaskAttemptID(
      boolean isMap, boolean isSpeculative) {
      JobID jobId = getJobID();
      TaskType t = TaskType.REDUCE;
      if (isMap) {
        t = TaskType.MAP;
      }
      if (!isSpeculative) {
        return new TaskAttemptID(
          jobId.getJtIdentifier(), jobId.getId(), t,
          (isMap) ? ++mapTaskCtr : ++redTaskCtr, 0);
      } else {
        return new TaskAttemptID(
          jobId.getJtIdentifier(), jobId.getId(), t,
          (isMap) ? mapTaskCtr : redTaskCtr, 1);
      }
    }

    @Override
    Set<TaskInProgress> getNonLocalRunningMaps() {
      return (Set<TaskInProgress>) mapTips;
    }

    @Override
    Set<TaskInProgress> getRunningReduces() {
      return (Set<TaskInProgress>) reduceTips;
    }

  }

  static class FakeFailingJobInProgress extends FakeJobInProgress {

    public FakeFailingJobInProgress(
      JobID id, JobConf jobConf,
      FakeTaskTrackerManager taskTrackerManager, String user, 
      JobTracker jt) throws IOException {
      super(id, jobConf, taskTrackerManager, user, jt);
    }

    @Override
    public synchronized void initTasks() throws IOException {
      throw new IOException("Failed Initalization");
    }

    @Override
    synchronized void fail() {
      this.status.setRunState(JobStatus.FAILED);
    }
  }

  static class FakeTaskInProgress extends TaskInProgress {
    private boolean isMap;
    private FakeJobInProgress fakeJob;
    private TreeMap<TaskAttemptID, String> activeTasks;
    private TaskStatus taskStatus;
    boolean hasSpeculativeMap;
    boolean hasSpeculativeReduce;

    FakeTaskInProgress(
      JobID jId, JobConf jobConf, Task t,
      boolean isMap, FakeJobInProgress job, 
      JobSplit.TaskSplitMetaInfo split) {
      super(jId, "", split, null, jobConf, job, 0, 1);
      this.isMap = isMap;
      this.fakeJob = job;
      activeTasks = new TreeMap<TaskAttemptID, String>();
      activeTasks.put(t.getTaskID(), "tt");
      // create a fake status for a task that is running for a bit
      this.taskStatus = TaskStatus.createTaskStatus(isMap);
      taskStatus.setProgress(0.5f);
      taskStatus.setRunState(TaskStatus.State.RUNNING);
      if (jobConf.getMapSpeculativeExecution()) {
        //resetting of the hasSpeculativeMap is done
        //when speculative map is scheduled by the job.
        hasSpeculativeMap = true;
      }
      if (jobConf.getReduceSpeculativeExecution()) {
        //resetting of the hasSpeculativeReduce is done
        //when speculative reduce is scheduled by the job.
        hasSpeculativeReduce = true;
      }
    }

    @Override
    TreeMap<TaskAttemptID, String> getActiveTasks() {
      return activeTasks;
    }

    @Override
    public TaskStatus getTaskStatus(TaskAttemptID taskid) {
      // return a status for a task that has run a bit
      return taskStatus;
    }

    @Override
    boolean killTask(TaskAttemptID taskId, boolean shouldFail) {
      if (isMap) {
        fakeJob.mapTaskFinished();
      } else {
        fakeJob.reduceTaskFinished();
      }
      return true;
    }

    @Override
      /*
      *hasSpeculativeMap and hasSpeculativeReduce is reset by FakeJobInProgress
      *after the speculative tip has been scheduled.
      */
    boolean canBeSpeculated(long currentTime) {
      if (isMap && hasSpeculativeMap) {
        return fakeJob.getJobConf().getMapSpeculativeExecution();
      }
      if (!isMap && hasSpeculativeReduce) {
        return fakeJob.getJobConf().getReduceSpeculativeExecution();
      }
      return false;
    }

    @Override
    public boolean isRunning() {
      return !activeTasks.isEmpty();
    }

  }

  static class FakeQueueManager extends QueueManager {
    private Set<String> queueNames = null;
    private static final AccessControlList allEnabledAcl =
      new AccessControlList("*");

    FakeQueueManager(Configuration conf) {
      super(conf);
    }

    void setQueues(Set<String> queueNames) {
      this.queueNames = queueNames;

      // sync up queues with the parent class.
      Queue[] queues = new Queue[queueNames.size()];
      int i = 0;
      for (String queueName : queueNames) {
        HashMap<String, AccessControlList> aclsMap
          = new HashMap<String, AccessControlList>();
        for (QueueACL qAcl : QueueACL.values()) {
          String key = toFullPropertyName(queueName, qAcl.getAclName());
          aclsMap.put(key, allEnabledAcl);
        }
        queues[i++] = new Queue(queueName, aclsMap, QueueState.RUNNING);
      }
      super.setQueues(queues);
    }

    public synchronized Set<String> getLeafQueueNames() {
      return queueNames;
    }

  }

  static class FakeTaskTrackerManager implements TaskTrackerManager {
    int maps = 0;
    int reduces = 0;
    int maxMapTasksPerTracker = 2;
    int maxReduceTasksPerTracker = 1;
    long ttExpiryInterval = 10 * 60 * 1000L; // default interval
    List<JobInProgressListener> listeners =
      new ArrayList<JobInProgressListener>();

    QueueManager qm = null;

    private Map<String, TaskTracker> trackers =
      new HashMap<String, TaskTracker>();
    private Map<String, TaskStatus> taskStatuses =
      new HashMap<String, TaskStatus>();
    private Map<JobID, JobInProgress> jobs =
      new HashMap<JobID, JobInProgress>();
    protected JobConf defaultJobConf;
    private int jobCounter = 0;

    public FakeTaskTrackerManager() {
      this(2, 2, 1);
    }

    public FakeTaskTrackerManager(
      int numTaskTrackers,
      int maxMapTasksPerTracker, int maxReduceTasksPerTracker) {
    Configuration cfn = new Configuration();
    cfn.set("mapred.queue.names","default");
    qm = new FakeQueueManager(cfn);
      this.maxMapTasksPerTracker = maxMapTasksPerTracker;
      this.maxReduceTasksPerTracker = maxReduceTasksPerTracker;
      for (int i = 1; i < numTaskTrackers + 1; i++) {
        String ttName = "tt" + i;
        TaskTracker tt = new TaskTracker(ttName);
        tt.setStatus(
          new TaskTrackerStatus(
            ttName, ttName + ".host", i,
            new ArrayList<TaskStatus>(), 0,
            maxMapTasksPerTracker,
            maxReduceTasksPerTracker));
        trackers.put(ttName, tt);
      }

      defaultJobConf = new JobConf();
      defaultJobConf.set("mapred.queue.names","default");
      //by default disable speculative execution.
      defaultJobConf.setMapSpeculativeExecution(false);
      defaultJobConf.setReduceSpeculativeExecution(false);
    }

    public void addTaskTracker(String ttName) {
      TaskTracker tt = new TaskTracker(ttName);
      tt.setStatus(
        new TaskTrackerStatus(
          ttName, ttName + ".host", 1,
          new ArrayList<TaskStatus>(), 0,
          maxMapTasksPerTracker,
          maxReduceTasksPerTracker));
      trackers.put(ttName, tt);
    }

    public ClusterStatus getClusterStatus() {
      int numTrackers = trackers.size();
      return new ClusterStatus(
        numTrackers, 0,
        ttExpiryInterval, maps, reduces,
        numTrackers * maxMapTasksPerTracker,
        numTrackers * maxReduceTasksPerTracker,
        JobTrackerStatus.RUNNING);
    }

    public int getNumberOfUniqueHosts() {
      return 0;
    }

    public int getNextHeartbeatInterval() {
      return JTConfig.JT_HEARTBEAT_INTERVAL_MIN_DEFAULT;
    }

    /**
     * Kill Job, and all its tasks on the corresponding TaskTrackers.
     */
    @Override
    public void killJob(JobID jobid) throws IOException {
      killJob(jobid, true);
    }

    /**
     * Kill the job, and kill the tasks on the corresponding TaskTrackers only
     * if killTasks is true.
     * 
     * @param jobid
     * @throws IOException
     */
    void killJob(JobID jobid, boolean killTasks)
        throws IOException {
      JobInProgress job = jobs.get(jobid);
      // Finish all the tasks for this job
      if (job instanceof FakeJobInProgress && killTasks) {
        FakeJobInProgress fJob = (FakeJobInProgress) job;
        for (String tipID : taskStatuses.keySet()) {
          finishTask(tipID, fJob, TaskStatus.State.KILLED);
        }
      }
      finalizeJob(job, JobStatus.KILLED);
      job.kill();
    }

    public void initJob(JobInProgress jip) {
      try {
        JobStatus oldStatus = (JobStatus) jip.getStatus().clone();
        jip.initTasks();
        if (jip.isJobEmpty()) {
          completeEmptyJob(jip);
        } else if (!jip.isSetupCleanupRequired()) {
          jip.completeSetup();
        }
        JobStatus newStatus = (JobStatus) jip.getStatus().clone();
        JobStatusChangeEvent event = new JobStatusChangeEvent(
          jip,
          JobStatusChangeEvent.EventType.RUN_STATE_CHANGED, oldStatus,
          newStatus);
        for (JobInProgressListener listener : listeners) {
          listener.jobUpdated(event);
        }
      } catch (Exception ioe) {
        failJob(jip);
      }
    }

    private synchronized void completeEmptyJob(JobInProgress jip) {
      jip.completeEmptyJob();
    }

    public synchronized void failJob(JobInProgress jip) {
      JobStatus oldStatus = (JobStatus) jip.getStatus().clone();
      jip.fail();
      JobStatus newStatus = (JobStatus) jip.getStatus().clone();
      JobStatusChangeEvent event = new JobStatusChangeEvent(
        jip,
        JobStatusChangeEvent.EventType.RUN_STATE_CHANGED, oldStatus, newStatus);
      for (JobInProgressListener listener : listeners) {
        listener.jobUpdated(event);
      }
    }

    public void retireJob(JobID jobid) {
      jobs.remove(jobid);
    }

    @Override
    public JobInProgress getJob(JobID jobid) {
      return jobs.get(jobid);
    }

    Collection<JobInProgress> getJobs() {
      return jobs.values();
    }

    public Collection<TaskTrackerStatus> taskTrackers() {
      List<TaskTrackerStatus> statuses = new ArrayList<TaskTrackerStatus>();
      for (TaskTracker tt : trackers.values()) {
        statuses.add(tt.getStatus());
      }
      return statuses;
    }


    public void addJobInProgressListener(JobInProgressListener listener) {
      listeners.add(listener);
    }

    public void removeJobInProgressListener(JobInProgressListener listener) {
      listeners.remove(listener);
    }

    public void submitJob(JobInProgress job) throws IOException {
      jobs.put(job.getJobID(), job);
      for (JobInProgressListener listener : listeners) {
        listener.jobAdded(job);
      }
    }

    FakeJobInProgress submitJob(int state, JobConf jobConf)
        throws IOException {
      FakeJobInProgress job =
          new FakeJobInProgress(new JobID("test", ++jobCounter),
              (jobConf == null ? new JobConf(defaultJobConf) : jobConf), this,
              jobConf.getUser(), UtilsForTests.getJobTracker());
      job.getStatus().setRunState(state);
      this.submitJob(job);
      return job;
    }

    FakeJobInProgress submitJobAndInit(int state, JobConf jobConf)
        throws IOException {
      FakeJobInProgress j = submitJob(state, jobConf);
      this.initJob(j);
      return j;
    }

    FakeJobInProgress submitJob(int state, int maps, int reduces,
        String queue, String user)
        throws IOException {
      JobConf jobConf = new JobConf(defaultJobConf);
      jobConf.setNumMapTasks(maps);
      jobConf.setNumReduceTasks(reduces);
      if (queue != null) {
        jobConf.setQueueName(queue);
      }
      jobConf.setUser(user);
      return submitJob(state, jobConf);
    }

    // Submit a job and update the listeners
    FakeJobInProgress submitJobAndInit(int state, int maps, int reduces,
        String queue, String user)
        throws IOException {
      FakeJobInProgress j = submitJob(state, maps, reduces, queue, user);
      this.initJob(j);
      return j;
    }

    HashMap<String, ArrayList<FakeJobInProgress>> submitJobs(
        int numberOfUsers, int numberOfJobsPerUser, String queue)
        throws Exception {
      HashMap<String, ArrayList<FakeJobInProgress>> userJobs =
          new HashMap<String, ArrayList<FakeJobInProgress>>();
      for (int i = 1; i <= numberOfUsers; i++) {
        String user = String.valueOf("u" + i);
        ArrayList<FakeJobInProgress> jips = new ArrayList<FakeJobInProgress>();
        for (int j = 1; j <= numberOfJobsPerUser; j++) {
          jips.add(this.submitJob(JobStatus.PREP, 1, 1, queue, user));
        }
        userJobs.put(user, jips);
      }
      return userJobs;
    }

    public TaskTracker getTaskTracker(String trackerID) {
      return trackers.get(trackerID);
    }

    public void startTask(String taskTrackerName, final Task t) {
      if (t.isMapTask()) {
        maps++;
      } else {
        reduces++;
      }
      TaskStatus status = new TaskStatus() {
        @Override
        public TaskAttemptID getTaskID() {
          return t.getTaskID();
        }

        @Override
        public boolean getIsMap() {
          return t.isMapTask();
        }

        @Override
        public int getNumSlots() {
          return t.getNumSlotsRequired();
        }

        @Override
        public void addFetchFailedMap(TaskAttemptID mapTaskId) {
          
        }
      };
      taskStatuses.put(t.getTaskID().toString(), status);
      status.setRunState(TaskStatus.State.RUNNING);
      trackers.get(taskTrackerName).getStatus().getTaskReports().add(status);
    }

    public void finishTask(
        String tipId,
        FakeJobInProgress j) {
      finishTask(tipId, j, TaskStatus.State.SUCCEEDED);
    }

    public void finishTask(String tipId, FakeJobInProgress j,
        TaskStatus.State taskState) {
      TaskStatus status = taskStatuses.get(tipId);
      if (status.getIsMap()) {
        maps--;
        j.mapTaskFinished();
      } else {
        reduces--;
        j.reduceTaskFinished();
      }
      status.setRunState(taskState);
    }

    void finalizeJob(FakeJobInProgress fjob) {
      finalizeJob(fjob, JobStatus.SUCCEEDED);
    }

    void finalizeJob(JobInProgress fjob, int state) {
      // take a snapshot of the status before changing it
      JobStatus oldStatus = (JobStatus) fjob.getStatus().clone();
      fjob.getStatus().setRunState(state);
      JobStatus newStatus = (JobStatus) fjob.getStatus().clone();
      JobStatusChangeEvent event =
        new JobStatusChangeEvent(
          fjob, JobStatusChangeEvent.EventType.RUN_STATE_CHANGED, oldStatus,
          newStatus);
      for (JobInProgressListener listener : listeners) {
        listener.jobUpdated(event);
      }
    }

    public void setPriority(FakeJobInProgress fjob, JobPriority priority) {
      // take a snapshot of the status before changing it
      JobStatus oldStatus = (JobStatus) fjob.getStatus().clone();
      fjob.setPriority(priority);
      JobStatus newStatus = (JobStatus) fjob.getStatus().clone();
      JobStatusChangeEvent event =
        new JobStatusChangeEvent(
          fjob, JobStatusChangeEvent.EventType.PRIORITY_CHANGED, oldStatus,
          newStatus);
      for (JobInProgressListener listener : listeners) {
        listener.jobUpdated(event);
      }
    }

    public void setStartTime(FakeJobInProgress fjob, long start) {
      // take a snapshot of the status before changing it
      JobStatus oldStatus = (JobStatus) fjob.getStatus().clone();

      fjob.startTime = start; // change the start time of the job
      fjob.status.setStartTime(start); // change the start time of the jobstatus

      JobStatus newStatus = (JobStatus) fjob.getStatus().clone();

      JobStatusChangeEvent event =
        new JobStatusChangeEvent(
          fjob, JobStatusChangeEvent.EventType.START_TIME_CHANGED, oldStatus,
          newStatus);
      for (JobInProgressListener listener : listeners) {
        listener.jobUpdated(event);
      }
    }

    void addQueues(String[] arr) {
      Set<String> queues = new HashSet<String>();
      for (String s : arr) {
        queues.add(s);
      }
      ((FakeQueueManager)qm).setQueues(queues);
    }

    void setFakeQueues(List<CapacityTestUtils.FakeQueueInfo> queues) {
      for (CapacityTestUtils.FakeQueueInfo q : queues) {
        Properties p = new Properties();
        p.setProperty(CapacitySchedulerConf.CAPACITY_PROPERTY,
            String.valueOf(q.capacity));
        p.setProperty(CapacitySchedulerConf.MAX_CAPACITY_PROPERTY,
            String.valueOf(q.maxCapacity));
        p.setProperty(CapacitySchedulerConf.SUPPORTS_PRIORITY_PROPERTY,
            String.valueOf(q.supportsPrio));
        p.setProperty(
            CapacitySchedulerConf.MINIMUM_USER_LIMIT_PERCENT_PROPERTY,
            String.valueOf(q.ulMin));
        ((FakeQueueManager) qm).getQueue(q.queueName).setProperties(p);
      }
    }

    void setQueueManager(QueueManager qManager) {
      this.qm = qManager;
    }

    public QueueManager getQueueManager() {
      return qm;
    }

    @Override
    public boolean killTask(TaskAttemptID taskid, boolean shouldFail) {
      return true;
    }

    public JobID getNextJobID() {
      return new JobID("test", ++jobCounter);
    }

  }// represents a fake queue configuration info

  static class FakeQueueInfo {
    String queueName;
    float capacity;
    float maxCapacity = -1.0f;
    boolean supportsPrio;
    int ulMin;

    public FakeQueueInfo(
      String queueName, float capacity, boolean supportsPrio, int ulMin) {
      this.queueName = queueName;
      this.capacity = capacity;
      this.supportsPrio = supportsPrio;
      this.ulMin = ulMin;
    }
  }

}