hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acmur...@apache.org
Subject svn commit: r1464947 - in /hadoop/common/branches/branch-1: ./ src/mapred/ src/mapred/org/apache/hadoop/mapred/ src/test/org/apache/hadoop/mapred/
Date Fri, 05 Apr 2013 12:24:54 GMT
Author: acmurthy
Date: Fri Apr  5 12:24:53 2013
New Revision: 1464947

URL: http://svn.apache.org/r1464947
Log:
MAPREDUCE-4824. Provide a mechanism for jobs to indicate they should not be recovered on JobTracker
restart. Contributed by Tom White and Arun C. Murthy.

Modified:
    hadoop/common/branches/branch-1/CHANGES.txt
    hadoop/common/branches/branch-1/src/mapred/mapred-default.xml
    hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobConf.java
    hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobTracker.java
    hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestRecoveryManager.java

Modified: hadoop/common/branches/branch-1/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/CHANGES.txt?rev=1464947&r1=1464946&r2=1464947&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/CHANGES.txt (original)
+++ hadoop/common/branches/branch-1/CHANGES.txt Fri Apr  5 12:24:53 2013
@@ -102,6 +102,9 @@ Release 1.2.0 - unreleased
     HDFS-4651. Offline Image Viewer backport to branch-1.
     (Chris Nauroth via suresh)
 
+    MAPREDUCE-4824. Provide a mechanism for jobs to indicate they should not
+    be recovered on JobTracker restart. (tomwhite & acmurthy via acmurthy) 
+
   IMPROVEMENTS
 
     HADOOP-9434. Backport HADOOP-9267: hadoop -h|-{0,2}help should print usage.

Modified: hadoop/common/branches/branch-1/src/mapred/mapred-default.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/mapred/mapred-default.xml?rev=1464947&r1=1464946&r2=1464947&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/mapred/mapred-default.xml (original)
+++ hadoop/common/branches/branch-1/src/mapred/mapred-default.xml Fri Apr  5 12:24:53 2013
@@ -289,6 +289,17 @@
 </property>
 
 <property>
+  <name>mapreduce.job.restart.recover</name>
+  <value>true</value>
+  <description>A per-job override for job recovery. If set to false for a
+      job then job recovery will not be attempted for that job upon restart
+      even if mapred.jobtracker.restart.recover is enabled. Defaults to true
+      so that jobs are recovered by default if
+      mapred.jobtracker.restart.recover is enabled.
+  </description>
+</property>
+
+<property>
   <name>mapred.jobtracker.job.history.block.size</name>
   <value>3145728</value>
   <description>The block size of the job history file. Since the job recovery

Modified: hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobConf.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobConf.java?rev=1464947&r1=1464946&r2=1464947&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobConf.java (original)
+++ hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobConf.java Fri Apr
 5 12:24:53 2013
@@ -327,6 +327,11 @@ public class JobConf extends Configurati
   public static final String WORKFLOW_ADJACENCY_PREFIX_PATTERN =
       "^mapreduce\\.workflow\\.adjacency\\..+";
 
+  public static final String MAPREDUCE_RECOVER_JOB = 
+      "mapreduce.job.restart.recover";
+
+  public static final boolean DEFAULT_MAPREDUCE_RECOVER_JOB = true; 
+
   private Credentials credentials = new Credentials();
   
   /**

Modified: hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobTracker.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobTracker.java?rev=1464947&r1=1464946&r2=1464947&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobTracker.java (original)
+++ hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/JobTracker.java Fri
Apr  5 12:24:53 2013
@@ -1428,13 +1428,19 @@ public class JobTracker implements MRCon
           // Re-submit job  
           final UserGroupInformation ugi = UserGroupInformation
               .createRemoteUser(token.getUser().toString());
-          ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
+          JobStatus status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
             public JobStatus run() throws IOException, InterruptedException {
               return submitJob(JobID.downgrade(token.getJobID()), token
                   .getJobSubmitDir().toString(), ugi, ts, true);
             }
           });
-          recovered++;
+          if (status == null) {
+            LOG.info("Job " + jobId + " was not recovered since it " +
+              "disabled recovery on restart (" + JobConf.MAPREDUCE_RECOVER_JOB +
+              " set to 'false').");
+          } else {
+            recovered++;
+          }
         } catch (Exception e) {
           LOG.warn("Could not recover job " + jobId, e);
         }
@@ -3515,8 +3521,10 @@ public class JobTracker implements MRCon
    * JobStatus. Those two sub-objects are sometimes shipped outside of the
    * JobTracker. But JobInProgress adds info that's useful for the JobTracker
    * alone.
+   * @return null if the job is being recovered but mapred.job.restart.recover
+   * is false.
    */
-  public JobStatus submitJob(JobID jobId, String jobSubmitDir,
+  JobStatus submitJob(JobID jobId, String jobSubmitDir,
       UserGroupInformation ugi, Credentials ts, boolean recovered)
       throws IOException {
     // Check for safe-mode
@@ -3544,6 +3552,13 @@ public class JobTracker implements MRCon
       throw new IOException(e);
     }
     
+    if (recovered && 
+        !job.getJobConf().getBoolean(
+            JobConf.MAPREDUCE_RECOVER_JOB, 
+            JobConf.DEFAULT_MAPREDUCE_RECOVER_JOB)) {
+      return null;
+    }
+    
     synchronized (this) {
       // check if queue is RUNNING
       String queue = job.getProfile().getQueueName();

Modified: hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestRecoveryManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestRecoveryManager.java?rev=1464947&r1=1464946&r2=1464947&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestRecoveryManager.java
(original)
+++ hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestRecoveryManager.java
Fri Apr  5 12:24:53 2013
@@ -202,8 +202,8 @@ public class TestRecoveryManager {
         new Path(TEST_DIR, "output3"), 2, 0, "test-resubmission", signalFile,
         signalFile);
 
-    JobClient jc = new JobClient(job1);
-    RunningJob rJob1 = jc.submitJob(job1);
+    JobClient jc1 = new JobClient(job1);
+    RunningJob rJob1 = jc1.submitJob(job1);
     LOG.info("Submitted first job " + rJob1.getID());
 
     while (rJob1.mapProgress() < 0.5f) {
@@ -211,6 +211,27 @@ public class TestRecoveryManager {
       UtilsForTests.waitFor(100);
     }
 
+    // now submit job2
+    JobConf job2 = mr.createJobConf();
+
+    String signalFile1 = new Path(TEST_DIR, "signal1").toString();
+    UtilsForTests.configureWaitingJobConf(job2, 
+        new Path(TEST_DIR, "input"), new Path(TEST_DIR, "output4"), 20, 0, 
+        "test-recovery-manager", signalFile1, signalFile1);
+    job2.setBoolean(JobConf.MAPREDUCE_RECOVER_JOB, false); // don't recover
+    
+    // submit the job
+    RunningJob rJob2 = (new JobClient(job2)).submitJob(job2);
+    LOG.info("Submitted job " + rJob2.getID());
+    
+    // wait for it to init
+    JobInProgress jip2 = jobtracker.getJob(rJob2.getID());
+    
+    while (!jip2.inited()) {
+      LOG.info("Waiting for job " + jip2.getJobID() + " to be inited");
+      UtilsForTests.waitFor(100);
+    }
+
     // kill the jobtracker
     LOG.info("Stopping jobtracker");
     mr.stopJobTracker();
@@ -218,11 +239,11 @@ public class TestRecoveryManager {
     // start the jobtracker
     LOG.info("Starting jobtracker");
     mr.startJobTracker();
-    UtilsForTests.waitForJobTracker(jc);
+    UtilsForTests.waitForJobTracker(jc1);
 
     jobtracker = mr.getJobTrackerRunner().getJobTracker();
 
-    // assert that job is recovered by the jobtracker
+    // assert that only job1 is recovered by the jobtracker
     Assert.assertEquals("Resubmission failed ", 1, 
         jobtracker.getAllJobs().length);
 



Mime
View raw message