Merge -r 736069:736070 from trunk to branch 0.20 to fix HADOOP-5048.

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

CHANGES.txt

@@ -561,6 +561,10 @@ Release 0.20.0 - Unreleased
     HADOOP-4993. Fix Chukwa agent configuration and startup to make it both
     more modular and testable. (Ari Rabkin via cdouglas)
 
+    HADOOP-5048. Fix capacity scheduler to correctly cleanup jobs that are
+    killed after initialization, but before running. 
+    (Sreekanth Ramakrishnan via yhemanth)
+
 Release 0.19.1 - Unreleased
 
   IMPROVEMENTS

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

@@ -1276,7 +1276,7 @@ class CapacityTaskScheduler extends TaskScheduler {
       }
       
       //update stats on waiting jobs
-      for(JobInProgress j: jobQueuesManager.getJobs(qsi.queueName)) {
+      for(JobInProgress j: jobQueuesManager.getWaitingJobs(qsi.queueName)) {
         // pending tasks
         if ((qsi.mapTSI.numPendingTasks > mapClusterCapacity) &&
             (qsi.reduceTSI.numPendingTasks > reduceClusterCapacity)) {
@@ -1498,7 +1498,7 @@ class CapacityTaskScheduler extends TaskScheduler {
       jobCollection.addAll(runningJobs);
     }
     Collection<JobInProgress> waitingJobs = 
-      jobQueuesManager.getJobs(queueName);
+      jobQueuesManager.getWaitingJobs(queueName);
     Collection<JobInProgress> tempCollection = new ArrayList<JobInProgress>();
     if(waitingJobs != null) {
       tempCollection.addAll(waitingJobs);

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

@@ -243,7 +243,7 @@ public class JobInitializationPoller extends Thread {
    * Set of jobs which have been passed to Initialization threads.
    * This is maintained so that we dont call initTasks() for same job twice.
    */
-  private HashSet<JobID> initializedJobs;
+  private HashMap<JobID,JobInProgress> initializedJobs;
 
   private volatile boolean running;
 
@@ -255,7 +255,7 @@ public class JobInitializationPoller extends Thread {
 
   public JobInitializationPoller(JobQueuesManager mgr,
       CapacitySchedulerConf rmConf, Set<String> queue) {
-    initializedJobs = new HashSet<JobID>();
+    initializedJobs = new HashMap<JobID,JobInProgress>();
     jobQueues = new HashMap<String, QueueInfo>();
     this.jobQueueManager = mgr;
     threadsToQueueMap = new HashMap<String, JobInitializationThread>();
@@ -293,9 +293,20 @@ public class JobInitializationPoller extends Thread {
     }
   }
 
+  /**
+   * This is main thread of initialization poller, We essentially do 
+   * following in the main threads:
+   * 
+   * <ol>
+   * <li> Clean up the list of initialized jobs list which poller maintains
+   * </li>
+   * <li> Select jobs to initialize in the polling interval.</li>
+   * </ol>
+   */
   public void run() {
     while (running) {
       try {
+        cleanUpInitializedJobsList();
         selectJobsToInitialize();
         if (!this.isInterrupted()) {
           Thread.sleep(sleepInterval);
@@ -307,17 +318,18 @@ public class JobInitializationPoller extends Thread {
     }
   }
 
-  // The key method that picks up jobs to initialize for each queue.
-  // The jobs picked up are added to the worker thread that is handling
-  // initialization for that queue.
-  // The method is package private to allow tests to call it synchronously
-  // in a controlled manner.
+  /**
+   * The key method which does selecting jobs to be initalized across 
+   * queues and assign those jobs to their appropriate init-worker threads.
+   * <br/>
+   * This method is overriden in test case which is used to test job
+   * initialization poller.
+   * 
+   */
   void selectJobsToInitialize() {
     for (String queue : jobQueues.keySet()) {
       ArrayList<JobInProgress> jobsToInitialize = getJobsToInitialize(queue);
-      //if (LOG.isDebugEnabled()) {
-        printJobs(jobsToInitialize);
-      //}
+      printJobs(jobsToInitialize);
       JobInitializationThread t = threadsToQueueMap.get(queue);
       for (JobInProgress job : jobsToInitialize) {
         t.addJobsToQueue(queue, job);
@@ -325,6 +337,13 @@ public class JobInitializationPoller extends Thread {
     }
   }
 
+  /**
+   * Method used to print log statements about which jobs are being
+   * passed to init-threads. 
+   * 
+   * @param jobsToInitialize list of jobs which are passed to be 
+   * init-threads.
+   */
   private void printJobs(ArrayList<JobInProgress> jobsToInitialize) {
     for (JobInProgress job : jobsToInitialize) {
       LOG.info("Passing to Initializer Job Id :" + job.getJobID()
@@ -333,13 +352,25 @@ public class JobInitializationPoller extends Thread {
     }
   }
 
-  // This method exists to be overridden by test cases that wish to
-  // create a test-friendly worker thread which can be controlled
-  // synchronously.
+  /**
+   * This method exists to be overridden by test cases that wish to
+   * create a test-friendly worker thread which can be controlled
+   * synchronously.
+   * 
+   * @return Instance of worker init-threads.
+   */
   JobInitializationThread createJobInitializationThread() {
     return new JobInitializationThread();
   }
   
+  /**
+   * Method which is used by the poller to assign appropriate worker thread
+   * to a queue. The number of threads would be always less than or equal
+   * to number of queues in a system. If number of threads is configured to 
+   * be more than number of queues then poller does not create threads more
+   * than number of queues. 
+   * 
+   */
   private void assignThreadsToQueues() {
     int countOfQueues = jobQueues.size();
     String[] queues = (String[]) jobQueues.keySet().toArray(
@@ -369,14 +400,38 @@ public class JobInitializationPoller extends Thread {
     }
   }
 
-  /*
-   * Select jobs to be initialized for a given queue.
+  /**
+   * 
+   * Method used to select jobs to be initialized for a given queue. <br/>
+   * 
+   * We want to ensure that enough jobs have been initialized, so that when the
+   * Scheduler wants to consider a new job to run, it's ready. We clearly don't
+   * want to initialize too many jobs as each initialized job has a memory
+   * footprint, sometimes significant.
+   * 
+   * Number of jobs to be initialized is restricted by two values: - Maximum
+   * number of users whose jobs we want to initialize, which is equal to 
+   * the number of concurrent users the queue can support. - Maximum number 
+   * of initialized jobs per user. The product of these two gives us the
+   * total number of initialized jobs.
    * 
-   * The jobs are selected such that they are within the limits
-   * for number of users and number of jobs per user in the queue.
-   * The only exception is if high priority jobs are waiting to be
-   * initialized. In that case, we could exceed the configured limits.
-   * However, we try to restrict the excess to a minimum.
+   * Note that this is a rough number, meant for decreasing extra memory
+   * footprint. It's OK if we go over it once in a while, if we have to.
+   * 
+   * This can happen as follows. Suppose we have initialized 3 jobs for a
+   * user. Now, suppose the user submits a job who's priority is higher than
+   * that of the 3 jobs initialized. This job needs to be initialized, since it
+   * will run earlier than the 3 jobs. We'll now have 4 initialized jobs for the
+   * user. If memory becomes a problem, we should ideally un-initialize one of
+   * the 3 jobs, to keep the count of initialized jobs at 3, but that's
+   * something we don't do for now. This situation can also arise when a new
+   * user submits a high priority job, thus superceeding a user whose jobs have
+   * already been initialized. The latter user's initialized jobs are redundant,
+   * but we'll leave them initialized.
+   * 
+   * @param queue name of the queue to pick the jobs to initialize.
+   * @return list of jobs to be initalized in a queue. An empty queue is
+   *         returned if no jobs are found.
    */
   ArrayList<JobInProgress> getJobsToInitialize(String queue) {
     QueueInfo qi = jobQueues.get(queue);
@@ -385,116 +440,131 @@ public class JobInitializationPoller extends Thread {
     // queue.
     int maximumUsersAllowedToInitialize = qi.maxUsersAllowedToInitialize;
     int maxJobsPerUserAllowedToInitialize = qi.maxJobsPerUserToInitialize;
-    // calculate maximum number of jobs which can be allowed to initialize
-    // for this queue.
-    // This value is used when a user submits a high priority job after we
-    // have initialized jobs for that queue and none of them is scheduled.
-    // This would prevent us from initializing extra jobs for that particular
-    // user. Explanation given at end of method.
     int maxJobsPerQueueToInitialize = maximumUsersAllowedToInitialize
         * maxJobsPerUserAllowedToInitialize;
-    Collection<JobInProgress> jobs = jobQueueManager.getJobs(queue);
     int countOfJobsInitialized = 0;
     HashMap<String, Integer> userJobsInitialized = new HashMap<String, Integer>();
+    Collection<JobInProgress> jobs = jobQueueManager.getWaitingJobs(queue);
+    /*
+     * Walk through the collection of waiting jobs.
+     *  We maintain a map of jobs that have already been initialized. If a 
+     *  job exists in that map, increment the count for that job's user 
+     *  and move on to the next job.
+     *   
+     *  If the job doesn't exist, see whether we  want to initialize it. 
+     *  We initialize it if: - at least one job of the user has already 
+     *  been initialized, but the user's total initialized jobs are below 
+     *  the limit, OR - this is a new user, and we haven't reached the limit
+     *  for the number of users whose jobs we want to initialize. We break 
+     *  when we've reached the limit of maximum jobs to initialize.
+     */
     for (JobInProgress job : jobs) {
-      /*
-       * First check if job has been scheduled or completed or killed. If so
-       * then remove from uninitialised jobs. Remove from Job queue
-       */
-      if ((job.getStatus().getRunState() == JobStatus.RUNNING)
-          && (job.runningMaps() > 0 || job.runningReduces() > 0
-              || job.finishedMaps() > 0 || job.finishedReduces() > 0)) {
-        LOG.debug("Removing from the queue " + job.getJobID());
-        initializedJobs.remove(job.getJobID());
-        jobQueueManager.removeJobFromQueue(job);
-        continue;
-      } else if (job.isComplete()) {
-        LOG.debug("Removing from completed job from " + "the queue "
-            + job.getJobID());
-        initializedJobs.remove(job.getJobID());
-        jobQueueManager.removeJobFromQueue(job);
-        continue;
-      }
       String user = job.getProfile().getUser();
       int numberOfJobs = userJobsInitialized.get(user) == null ? 0
           : userJobsInitialized.get(user);
       // If the job is already initialized then add the count against user
       // then continue.
-      if (initializedJobs.contains(job.getJobID())) {
+      if (initializedJobs.containsKey(job.getJobID())) {
         userJobsInitialized.put(user, Integer.valueOf(numberOfJobs + 1));
         countOfJobsInitialized++;
         continue;
       }
       boolean isUserPresent = userJobsInitialized.containsKey(user);
-      /*
-       * If the user is present in user list and size of user list is less
-       * maximum allowed users initialize then initialize this job and add this
-       * user to the global list.
-       * 
-       * Else if he is present we check if his number of jobs has not crossed
-       * his quota and global quota.
-       * 
-       * The logic behind using a global per queue job can be understood by example
-       * below: Consider 3 users submitting normal priority job in a job queue with
-       * user limit as 100. (Max jobs per user = 2)
-       * 
-       * U1J1,U1J2,U1J3....,U3J3.
-       * 
-       * Jobs initialized would be
-       * 
-       * U1J1,U1J2,U2J1,U2J2,U3J1,U3J2
-       * 
-       * Now consider a case where U4 comes in and submits a high priority job.
-       * 
-       * U4J1 --- High Priority JOb, U4J2---- Normal priority job.
-       * 
-       * So, if we dont use global per queue value we would end up initializing both
-       * U4 jobs which is not correct.
-       * 
-       * By using a global value we ensure that we dont initialize any extra jobs
-       * for a user.
-       */
       if (!isUserPresent
           && userJobsInitialized.size() < maximumUsersAllowedToInitialize) {
         // this is a new user being considered and the number of users
         // is within limits.
         userJobsInitialized.put(user, Integer.valueOf(numberOfJobs + 1));
         jobsToInitialize.add(job);
-        initializedJobs.add(job.getJobID());
+        initializedJobs.put(job.getJobID(),job);
         countOfJobsInitialized++;
       } else if (isUserPresent
-          && numberOfJobs < maxJobsPerUserAllowedToInitialize
-          && countOfJobsInitialized < maxJobsPerQueueToInitialize) {
-        /*
-         * this is an existing user and the number of jobs per user
-         * is within limits, as also the number of jobs per queue.
-         * We need the check on number of jobs per queue to restrict
-         * the number of jobs we initialize over the limit due to high
-         * priority jobs.
-         * 
-         * For e.g Consider 3 users submitting normal priority job in 
-         * a job queue with user limit as 100 and max jobs per user as 2
-         * Say the jobs are U1J1,U1J2,U1J3....,U3J3.
-         * 
-         * Jobs initialized would be U1J1,U1J2,U2J1,U2J2,U3J1,U3J2
-         * 
-         * Now consider a case where U4 comes in and submits a high priority job
-         * and a normal priority job. Say U4J1 and U4J2
-         * 
-         * If we dont consider the number of jobs per queue we would end up 
-         * initializing both jobs from U4. Initializing the second job is
-         * unnecessary.
-         */
+          && numberOfJobs < maxJobsPerUserAllowedToInitialize) {
         userJobsInitialized.put(user, Integer.valueOf(numberOfJobs + 1));
         jobsToInitialize.add(job);
-        initializedJobs.add(job.getJobID());
+        initializedJobs.put(job.getJobID(),job);
         countOfJobsInitialized++;
       }
+      /*
+       * if the maximum number of jobs to initalize for a queue is reached
+       * then we stop looking at further jobs. The jobs beyond this number
+       * can be initialized.
+       */
+      if(countOfJobsInitialized > maxJobsPerQueueToInitialize) {
+        break;
+      }
     }
     return jobsToInitialize;
   }
 
 
+  /**
+   * Method which is used internally to clean up the initialized jobs
+   * data structure which the job initialization poller uses to check
+   * if a job is initalized or not.
+   * 
+   * Algorithm for cleaning up task is as follows:
+   * 
+   * <ul>
+   * <li> For jobs in <b>initalizedJobs</b> list </li>
+   * <ul>
+   * <li> If job is running</li>
+   * <ul>
+   * <li> If job is scheduled then remove the job from the waiting queue 
+   * of the scheduler and <b>initalizedJobs</b>.<br/>
+   *  The check for a job is scheduled or not is done by following 
+   *  formulae:<br/> 
+   *  if pending <i>task</i> &lt; desired <i>task</i> then scheduled else
+   *  not scheduled.<br/>
+   *  The formulae would return <i>scheduled</i> if one task has run or failed,
+   *  any cases in which there has been a failure but not enough to mark task 
+   *  as failed, we return <i>not scheduled</i> in formulae.
+   * </li>
+   * </ul>
+   * 
+   * <li> If job is complete, then remove the job from <b>initalizedJobs</b>.
+   * </li>
+   * 
+   * </ul>
+   * </ul>
+   * 
+   */
+  void cleanUpInitializedJobsList() {
+    Iterator<Entry<JobID, JobInProgress>> jobsIterator = 
+      initializedJobs.entrySet().iterator();
+    while(jobsIterator.hasNext()) {
+      Entry<JobID,JobInProgress> entry = jobsIterator.next();
+      JobInProgress job = entry.getValue();
+      if (job.getStatus().getRunState() == JobStatus.RUNNING) {
+        if (isScheduled(job)) {
+          LOG.info("Removing scheduled jobs from waiting queue"
+              + job.getJobID());
+          jobsIterator.remove();
+          jobQueueManager.removeJobFromWaitingQueue(job);
+          continue;
+        }
+      }
+      if(job.isComplete()) {
+        LOG.info("Removing killed/completed job from initalized jobs " +
+        		"list : "+ job.getJobID());
+        jobsIterator.remove();
+      }
+    }
+  }
+
+  /**
+   * Convenience method to check if job has been scheduled or not.
+   * 
+   * The method may return false in case of job which has failure but
+   * has not failed the tip.
+   * @param job
+   * @return
+   */
+  private boolean isScheduled(JobInProgress job) {
+    return ((job.pendingMaps() < job.desiredMaps()) 
+        || (job.pendingReduces() < job.desiredReduces()));
+  }
+
   void terminate() {
     running = false;
     for (Entry<String, JobInitializationThread> entry : threadsToQueueMap
@@ -519,7 +589,7 @@ public class JobInitializationPoller extends Thread {
     }
   }
 
-  HashSet<JobID> getInitializedJobList() {
-    return initializedJobs;
+  Set<JobID> getInitializedJobList() {
+    return initializedJobs.keySet();
   }
 }

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

@@ -65,7 +65,7 @@ class JobQueuesManager extends JobInProgressListener {
 
     // whether the queue supports priorities
     boolean supportsPriorities;
-    Map<JobSchedulingInfo, JobInProgress> jobList; // for waiting jobs
+    Map<JobSchedulingInfo, JobInProgress> waitingJobs; // for waiting jobs
     Map<JobSchedulingInfo, JobInProgress> runningJobs; // for running jobs
     
     public Comparator<JobSchedulingInfo> comparator;
@@ -79,14 +79,14 @@ class JobQueuesManager extends JobInProgressListener {
       else {
         comparator = STARTTIME_JOB_COMPARATOR;
       }
-      jobList = new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
+      waitingJobs = new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
       runningJobs = new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
     }
     
-    Collection<JobInProgress> getJobs() {
-      synchronized (jobList) {
+    Collection<JobInProgress> getWaitingJobs() {
+      synchronized (waitingJobs) {
         return Collections.unmodifiableCollection(
-            new LinkedList<JobInProgress>(jobList.values()));
+            new LinkedList<JobInProgress>(waitingJobs.values()));
       }
     }
     
@@ -109,21 +109,21 @@ class JobQueuesManager extends JobInProgressListener {
       }
     }
     
-    JobInProgress removeJob(JobSchedulingInfo schedInfo) {
-      synchronized (jobList) {
-        return jobList.remove(schedInfo);
+    JobInProgress removeWaitingJob(JobSchedulingInfo schedInfo) {
+      synchronized (waitingJobs) {
+        return waitingJobs.remove(schedInfo);
       }
     }
     
-    void addJob(JobInProgress job) {
-      synchronized (jobList) {
-        jobList.put(new JobSchedulingInfo(job), job);
+    void addWaitingJob(JobInProgress job) {
+      synchronized (waitingJobs) {
+        waitingJobs.put(new JobSchedulingInfo(job), job);
       }
     }
     
     int getWaitingJobCount() {
-      synchronized (jobList) {
-       return jobList.size(); 
+      synchronized (waitingJobs) {
+       return waitingJobs.size(); 
       }
     }
     
@@ -157,11 +157,11 @@ class JobQueuesManager extends JobInProgressListener {
   }
   
   /**
-   * Returns the queue of Uninitialised jobs associated with queue name.
+   * Returns the queue of waiting jobs associated with queue name.
    * 
    */
-  public Collection<JobInProgress> getJobs(String queueName) {
-    return jobQueues.get(queueName).getJobs();
+  Collection<JobInProgress> getWaitingJobs(String queueName) {
+    return jobQueues.get(queueName).getWaitingJobs();
   }
   
   @Override
@@ -178,21 +178,23 @@ class JobQueuesManager extends JobInProgressListener {
     }
     // add job to waiting queue. It will end up in the right place, 
     // based on priority. 
-    qi.addJob(job);
+    qi.addWaitingJob(job);
     // let scheduler know. 
     scheduler.jobAdded(job);
   }
 
   /*
-   * The removal of the running jobs alone is done by the JobQueueManager.
-   * The removal of the jobs in the job queue is taken care by the
-   * JobInitializationPoller.
+   * Method removes the jobs from both running and waiting job queue in 
+   * job queue manager.
    */
   private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
                             QueueInfo qi) {
     LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
         + job.getProfile().getQueueName() + " has completed");
+    //remove jobs from both queue's a job can be in
+    //running and waiting queue at the same time.
     qi.removeRunningJob(oldInfo);
+    qi.removeWaitingJob(oldInfo);
     // let scheduler know
     scheduler.jobCompleted(job);
   }
@@ -206,8 +208,8 @@ class JobQueuesManager extends JobInProgressListener {
   private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo, 
                            QueueInfo qi) {
     
-    if(qi.removeJob(oldInfo) != null) {
-      qi.addJob(job);
+    if(qi.removeWaitingJob(oldInfo) != null) {
+      qi.addWaitingJob(job);
     }
     if(qi.removeRunningJob(oldInfo) != null) {
       qi.addRunningJob(job);
@@ -264,10 +266,10 @@ class JobQueuesManager extends JobInProgressListener {
     }
   }
   
-  void removeJobFromQueue(JobInProgress job) {
+  void removeJobFromWaitingQueue(JobInProgress job) {
     String queue = job.getProfile().getQueueName();
     QueueInfo qi = jobQueues.get(queue);
-    qi.removeJob(new JobSchedulingInfo(job));
+    qi.removeWaitingJob(new JobSchedulingInfo(job));
   }
   
   Comparator<JobSchedulingInfo> getComparator(String queue) {

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

@@ -128,6 +128,7 @@ public class TestCapacityScheduler extends TestCase {
 
     @Override
     void selectJobsToInitialize() {
+      super.cleanUpInitializedJobsList();
       super.selectJobsToInitialize();
       for (ControlledJobInitializer t : workers) {
         t.initializeJobs();
@@ -704,7 +705,7 @@ public class TestCapacityScheduler extends TestCase {
     // check if the jobs are missing from the waiting queue
     // The jobs are not removed from waiting queue until they are scheduled 
     assertEquals("Waiting queue is garbled on job init", 2, 
-                 scheduler.jobQueuesManager.getJobs("default")
+                 scheduler.jobQueuesManager.getWaitingJobs("default")
                           .size());
     
     // test if changing the job priority/start-time works as expected in the 
@@ -788,7 +789,7 @@ public class TestCapacityScheduler extends TestCase {
   private JobInProgress[] getJobsInQueue(boolean waiting) {
     Collection<JobInProgress> queue = 
       waiting 
-      ? scheduler.jobQueuesManager.getJobs("default")
+      ? scheduler.jobQueuesManager.getWaitingJobs("default")
       : scheduler.jobQueuesManager.getRunningJobQueue("default");
     return queue.toArray(new JobInProgress[0]);
   }
@@ -897,7 +898,7 @@ public class TestCapacityScheduler extends TestCase {
    
     JobQueuesManager mgr = scheduler.jobQueuesManager;
     
-    while(mgr.getJobs("default").size() < 4){
+    while(mgr.getWaitingJobs("default").size() < 4){
       Thread.sleep(1);
     }
     //Raise status change events for jobs submitted.
@@ -2147,10 +2148,10 @@ public class TestCapacityScheduler extends TestCase {
     
     // reference to the initializedJobs data structure
     // changes are reflected in the set as they are made by the poller
-    HashSet<JobID> initializedJobs = initPoller.getInitializedJobList();
+    Set<JobID> initializedJobs = initPoller.getInitializedJobList();
     
     // we should have 12 (3 x 4) jobs in the job queue
-    assertEquals(mgr.getJobs("default").size(), 12);
+    assertEquals(mgr.getWaitingJobs("default").size(), 12);
 
     // run one poller iteration.
     p.selectJobsToInitialize();
@@ -2264,7 +2265,7 @@ public class TestCapacityScheduler extends TestCase {
     scheduler.setInitializationPoller(p);
     scheduler.start();
     JobInitializationPoller initPoller = scheduler.getInitializationPoller();
-    HashSet<JobID> initializedJobsList = initPoller.getInitializedJobList();
+    Set<JobID> initializedJobsList = initPoller.getInitializedJobList();
 
     // submit 3 jobs for 3 users
     submitJobs(3,3,"default");
@@ -2317,22 +2318,38 @@ public class TestCapacityScheduler extends TestCase {
     scheduler.start();
     
     JobQueuesManager mgr = scheduler.jobQueuesManager;
-    JobInitializationPoller initPoller = scheduler.getInitializationPoller();
-    HashSet<JobID> initializedJobsList = initPoller.getInitializedJobList();
     
+    // check proper running job movement and completion
+    checkRunningJobMovementAndCompletion();
+
+    // check failed running job movement
+    checkFailedRunningJobMovement();
+
+    // Check job movement of failed initalized job
+    checkFailedInitializedJobMovement();
+
+    // Check failed waiting job movement
+    checkFailedWaitingJobMovement();
+    
+  }
+  
+  private void checkRunningJobMovementAndCompletion() throws IOException {
+    
+    JobQueuesManager mgr = scheduler.jobQueuesManager;
+    JobInitializationPoller p = scheduler.getInitializationPoller();
     // submit a job
     FakeJobInProgress job = 
       submitJob(JobStatus.PREP, 1, 1, "default", "u1");
     p.selectJobsToInitialize();
     
-    assertEquals(initializedJobsList.size(), 1);
+    assertEquals(p.getInitializedJobList().size(), 1);
 
     // make it running.
     raiseStatusChangeEvents(mgr);
     
     // it should be there in both the queues.
     assertTrue("Job not present in Job Queue",
-        mgr.getJobs("default").contains(job));
+        mgr.getWaitingJobs("default").contains(job));
     assertTrue("Job not present in Running Queue",
         mgr.getRunningJobQueue("default").contains(job));
     
@@ -2343,12 +2360,12 @@ public class TestCapacityScheduler extends TestCase {
     p.selectJobsToInitialize();
     
     // now this task should be removed from the initialized list.
-    assertTrue(initializedJobsList.isEmpty());
+    assertTrue(p.getInitializedJobList().isEmpty());
 
     // the job should also be removed from the job queue as tasks
     // are scheduled
     assertFalse("Job present in Job Queue",
-        mgr.getJobs("default").contains(job));
+        mgr.getWaitingJobs("default").contains(job));
     
     // complete tasks and job
     taskTrackerManager.finishTask("tt1", "attempt_test_0001_m_000001_0", job);
@@ -2360,8 +2377,78 @@ public class TestCapacityScheduler extends TestCase {
         mgr.getRunningJobQueue("default").contains(job));
   }
   
+  private void checkFailedRunningJobMovement() throws IOException {
+    
+    JobQueuesManager mgr = scheduler.jobQueuesManager;
+    
+    //submit a job and initalized the same
+    FakeJobInProgress job = 
+      submitJobAndInit(JobStatus.RUNNING, 1, 1, "default", "u1");
+    
+    //check if the job is present in running queue.
+    assertTrue("Running jobs list does not contain submitted job",
+        mgr.getRunningJobQueue("default").contains(job));
+    
+    taskTrackerManager.finalizeJob(job, JobStatus.KILLED);
+    
+    //check if the job is properly removed from running queue.
+    assertFalse("Running jobs list does not contain submitted job",
+        mgr.getRunningJobQueue("default").contains(job));
+    
+  }
+
+  private void checkFailedInitializedJobMovement() throws IOException {
+    
+    JobQueuesManager mgr = scheduler.jobQueuesManager;
+    JobInitializationPoller p = scheduler.getInitializationPoller();
+    
+    //submit a job
+    FakeJobInProgress job = submitJob(JobStatus.PREP, 1, 1, "default", "u1");
+    //Initialize the job
+    p.selectJobsToInitialize();
+    //Don't raise the status change event.
+    
+    //check in waiting and initialized jobs list.
+    assertTrue("Waiting jobs list does not contain the job",
+        mgr.getWaitingJobs("default").contains(job));
+    
+    assertTrue("Initialized job does not contain the job",
+        p.getInitializedJobList().contains(job.getJobID()));
+    
+    //fail the initalized job
+    taskTrackerManager.finalizeJob(job, JobStatus.KILLED);
+    
+    //Check if the job is present in waiting queue
+    assertFalse("Waiting jobs list contains failed job",
+        mgr.getWaitingJobs("default").contains(job));
+    
+    //run the poller to do the cleanup
+    p.selectJobsToInitialize();
+    
+    //check for failed job in the initialized job list
+    assertFalse("Initialized jobs  contains failed job",
+        p.getInitializedJobList().contains(job.getJobID()));
+  }
+  
+  private void checkFailedWaitingJobMovement() throws IOException {
+    JobQueuesManager mgr = scheduler.jobQueuesManager;
+    // submit a job
+    FakeJobInProgress job = submitJob(JobStatus.PREP, 1, 1, "default",
+        "u1");
+    
+    // check in waiting and initialized jobs list.
+    assertTrue("Waiting jobs list does not contain the job", mgr
+        .getWaitingJobs("default").contains(job));
+    // fail the waiting job
+    taskTrackerManager.finalizeJob(job, JobStatus.KILLED);
+
+    // Check if the job is present in waiting queue
+    assertFalse("Waiting jobs list contains failed job", mgr
+        .getWaitingJobs("default").contains(job));
+  }
+  
   private void raiseStatusChangeEvents(JobQueuesManager mgr) {
-    Collection<JobInProgress> jips = mgr.getJobs("default");
+    Collection<JobInProgress> jips = mgr.getWaitingJobs("default");
     for(JobInProgress jip : jips) {
       if(jip.getStatus().getRunState() == JobStatus.RUNNING) {
         JobStatusChangeEvent evt = new JobStatusChangeEvent(jip,