hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yhema...@apache.org
Subject svn commit: r772057 - in /hadoop/core/trunk: CHANGES.txt src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java
Date Wed, 06 May 2009 05:37:50 GMT
Author: yhemanth
Date: Wed May  6 05:37:50 2009
New Revision: 772057

URL: http://svn.apache.org/viewvc?rev=772057&view=rev
Log:
HADOOP-5719. Remove jobs that failed initialization from the waiting queue in the capacity
scheduler. Contributed by Sreekanth Ramakrishnan.

Modified:
    hadoop/core/trunk/CHANGES.txt
    hadoop/core/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
    hadoop/core/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java

Modified: hadoop/core/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=772057&r1=772056&r2=772057&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Wed May  6 05:37:50 2009
@@ -567,6 +567,9 @@
     HADOOP-5718. Remove the check for the default queue in capacity scheduler.
     (Sreekanth Ramakrishnan via yhemanth)
 
+    HADOOP-5719. Remove jobs that failed initialization from the waiting queue
+    in the capacity scheduler. (Sreekanth Ramakrishnan via yhemanth)
+
 Release 0.20.0 - Unreleased
 
   INCOMPATIBLE CHANGES

Modified: hadoop/core/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java?rev=772057&r1=772056&r2=772057&view=diff
==============================================================================
--- hadoop/core/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
(original)
+++ hadoop/core/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
Wed May  6 05:37:50 2009
@@ -20,7 +20,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
 import java.util.TreeMap;
@@ -149,8 +148,8 @@
           } catch (Throwable t) {
             LOG.info("Job initialization failed:\n"
                 + StringUtils.stringifyException(t));
-            if (job != null)
-              job.fail();
+            jobQueueManager.removeJobFromWaitingQueue(job);
+            job.fail(); 
           }
         }
       }

Modified: hadoop/core/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java?rev=772057&r1=772056&r2=772057&view=diff
==============================================================================
--- hadoop/core/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java
(original)
+++ hadoop/core/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java
Wed May  6 05:37:50 2009
@@ -37,7 +37,6 @@
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType;
 import org.apache.hadoop.mapreduce.TaskType;
-import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 
 
@@ -253,6 +252,24 @@
     
   }
   
+  static class FakeFailingJobInProgress extends FakeJobInProgress {
+
+    public FakeFailingJobInProgress(JobID id, JobConf jobConf,
+        FakeTaskTrackerManager taskTrackerManager, String user) {
+      super(id, jobConf, taskTrackerManager, user);
+    }
+    
+    @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;
@@ -2016,6 +2033,39 @@
     t = checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
   }
   
+  public void testFailedJobInitalizations() throws Exception {
+    String[] qs = {"default"};
+    taskTrackerManager.addQueues(qs);
+    ArrayList<FakeQueueInfo> queues = new ArrayList<FakeQueueInfo>();
+    queues.add(new FakeQueueInfo("default", 100.0f, true, 100));
+    resConf.setFakeQueues(queues);
+    scheduler.setResourceManagerConf(resConf);
+    scheduler.start();
+
+    JobQueuesManager mgr = scheduler.jobQueuesManager;
+    
+    //Submit a job whose initialization would fail always.
+    FakeJobInProgress job =
+      new FakeFailingJobInProgress(new JobID("test", ++jobCounter),
+          new JobConf(), taskTrackerManager,"u1");
+    job.getStatus().setRunState(JobStatus.PREP);
+    taskTrackerManager.submitJob(job);
+    //check if job is present in waiting list.
+    assertEquals("Waiting job list does not contain submitted job",
+        1, mgr.getWaitingJobCount("default"));
+    assertTrue("Waiting job does not contain submitted job", 
+        mgr.getWaitingJobs("default").contains(job));
+    //initialization should fail now.
+    controlledInitializationPoller.selectJobsToInitialize();
+    //Check if the job has been properly cleaned up.
+    assertEquals("Waiting job list contains submitted job",
+        0, mgr.getWaitingJobCount("default"));
+    assertFalse("Waiting job contains submitted job", 
+        mgr.getWaitingJobs("default").contains(job));
+    assertFalse("Waiting job contains submitted job", 
+        mgr.getRunningJobQueue("default").contains(job));
+  }
+
   private void checkRunningJobMovementAndCompletion() throws IOException {
     
     JobQueuesManager mgr = scheduler.jobQueuesManager;



Mime
View raw message