HADOOP-3864. Prevent the JobTracker from locking up when a job is being

initialized. (acmurthy via omalley)


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@684143 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -165,6 +165,9 @@ Trunk (unreleased changes)
     HADOOP-3863. Use a thread-local string encoder rather than a static one
     that is protected by a lock. (acmurthy via omalley)
 
+    HADOOP-3864. Prevent the JobTracker from locking up when a job is being
+    initialized. (acmurthy via omalley)
+
   BUG FIXES
 
     HADOOP-3563.  Refactor the distributed upgrade code so that it is 

src/mapred/org/apache/hadoop/mapred/JobInProgress.java

@@ -30,6 +30,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.Vector;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -122,7 +123,7 @@ class JobInProgress {
   long finishTime;
 
   private JobConf conf;
-  boolean tasksInited = false;
+  AtomicBoolean tasksInited = new AtomicBoolean(false);
 
   private LocalFileSystem localFs;
   private JobID jobId;
@@ -297,12 +298,22 @@ class JobInProgress {
     }
     return cache;
   }
+  
+  /**
+   * Check if the job has been initialized.
+   * @return <code>true</code> if the job has been initialized, 
+   *         <code>false</code> otherwise
+   */
+  public boolean inited() {
+    return tasksInited.get();
+  }
+  
   /**
    * Construct the splits, etc.  This is invoked from an async
    * thread so that split-computation doesn't block anyone.
    */
   public synchronized void initTasks() throws IOException {
-    if (tasksInited) {
+    if (tasksInited.get()) {
       return;
     }
 
@@ -341,7 +352,7 @@ class JobInProgress {
       status.setMapProgress(1.0f);
       status.setReduceProgress(1.0f);
       status.setRunState(JobStatus.SUCCEEDED);
-      tasksInited = true;
+      tasksInited.set(true);
       JobHistory.JobInfo.logStarted(profile.getJobID(), 
                                     System.currentTimeMillis(), 0, 0);
       JobHistory.JobInfo.logFinished(profile.getJobID(), 
@@ -375,7 +386,7 @@ class JobInProgress {
     }
 
     this.status = new JobStatus(status.getJobID(), 0.0f, 0.0f, JobStatus.RUNNING);
-    tasksInited = true;
+    tasksInited.set(true);
         
     JobHistory.JobInfo.logStarted(profile.getJobID(), System.currentTimeMillis(), numMapTasks, numReduceTasks);
   }
@@ -663,7 +674,7 @@ class JobInProgress {
                                             int clusterSize, 
                                             int numUniqueHosts
                                            ) throws IOException {
-    if (!tasksInited) {
+    if (!tasksInited.get()) {
       LOG.info("Cannot create task split for " + profile.getJobID());
       return null;
     }
@@ -698,7 +709,7 @@ class JobInProgress {
                                                int clusterSize,
                                                int numUniqueHosts
                                               ) throws IOException {
-    if (!tasksInited) {
+    if (!tasksInited.get()) {
       LOG.info("Cannot create task split for " + profile.getJobID());
       return null;
     }

src/mapred/org/apache/hadoop/mapred/JobTracker.java

@@ -1584,7 +1584,9 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     
   public synchronized void killJob(JobID jobid) {
     JobInProgress job = jobs.get(jobid);
-    job.kill();
+    if (job.inited()) {
+      job.kill();
+    }
   }
 
   public synchronized JobProfile getJobProfile(JobID jobid) {
@@ -1653,6 +1655,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     }
   }
     
+  TaskCompletionEvent[] EMPTY_EVENTS = new TaskCompletionEvent[0];
+  
   /* 
    * Returns a list of TaskCompletionEvent for the given job, 
    * starting from fromEventId.
@@ -1660,11 +1664,13 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
    */
   public synchronized TaskCompletionEvent[] getTaskCompletionEvents(
       JobID jobid, int fromEventId, int maxEvents) throws IOException{
-    TaskCompletionEvent[] events;
+    TaskCompletionEvent[] events = EMPTY_EVENTS;
 
     JobInProgress job = this.jobs.get(jobid);
     if (null != job) {
-      events = job.getTaskCompletionEvents(fromEventId, maxEvents);
+      if (job.inited()) {
+        events = job.getTaskCompletionEvents(fromEventId, maxEvents);
+      }
     }
     else {
       events = completedJobStatusStore.readJobTaskCompletionEvents(jobid, fromEventId, maxEvents);