MAPREDUCE-2535. Fix NPE in JobClient caused by retirement.

Contributed by Robert Joseph Evans


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

CHANGES.txt

@@ -47,6 +47,9 @@ Release 0.20.205.0 - unreleased
     MAPREDUCE-2558. Add JobTracker metrics for scheduling queues. (Jeffrey
     Naisbitt via cdouglas)
 
+    MAPREDUCE-2535. Fix NPE in JobClient caused by retirement. (Robert Joseph
+    Evans via cdouglas)
+
 Release 0.20.204.0 - unreleased
 
   NEW FEATURES

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

@@ -183,7 +183,8 @@ public class JobClient extends Configured implements MRConstants, Tool  {
    * a JobProfile object to provide some info, and interacts with the
    * remote service to provide certain functionality.
    */
-  class NetworkedJob implements RunningJob {
+  static class NetworkedJob implements RunningJob {
+    private JobSubmissionProtocol jobSubmitClient;
     JobProfile profile;
     JobStatus status;
     long statustime;
@@ -191,16 +192,26 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     /**
      * We store a JobProfile and a timestamp for when we last
      * acquired the job profile.  If the job is null, then we cannot
-     * perform any of the tasks.  The job might be null if the JobTracker
-     * has completely forgotten about the job.  (eg, 24 hours after the
-     * job completes.)
+     * perform any of the tasks, so we throw an exception.
+     * The job might be null if the JobTracker has completely forgotten
+     * about the job.  (eg, 24 hours after the job completes.)
      */
-    public NetworkedJob(JobStatus job) throws IOException {
+    public NetworkedJob(JobStatus job, JobProfile prof, JobSubmissionProtocol jobSubmitClient) throws IOException {
       this.status = job;
-      this.profile = jobSubmitClient.getJobProfile(job.getJobID());
+      this.profile = prof;
+      this.jobSubmitClient = jobSubmitClient;
+      if(this.status == null) {
+        throw new IOException("The Job status cannot be null");
+      }
+      if(this.profile == null) {
+        throw new IOException("The Job profile cannot be null");
+      }
+      if(this.jobSubmitClient == null) {
+        throw new IOException("The Job Submission Protocol cannot be null");
+      }
       this.statustime = System.currentTimeMillis();
     }
-
+    
     /**
      * Some methods rely on having a recent job profile object.  Refresh
      * it, if necessary
@@ -217,6 +228,9 @@ public class JobClient extends Configured implements MRConstants, Tool  {
      */
     synchronized void updateStatus() throws IOException {
       this.status = jobSubmitClient.getJobStatus(profile.getJobID());
+      if(this.status == null) {
+        throw new IOException("The job appears to have been removed."); 
+      }
       this.statustime = System.currentTimeMillis();
     }
 
@@ -863,8 +877,9 @@ public class JobClient extends Configured implements MRConstants, Tool  {
           printTokens(jobId, jobCopy.getCredentials());
           status = jobSubmitClient.submitJob(
               jobId, submitJobDir.toString(), jobCopy.getCredentials());
-          if (status != null) {
-            return new NetworkedJob(status);
+          JobProfile prof = jobSubmitClient.getJobProfile(jobId);
+          if (status != null && prof != null) {
+            return new NetworkedJob(status, prof, jobSubmitClient);
           } else {
             throw new IOException("Could not launch job");
           }
@@ -1011,8 +1026,9 @@ public class JobClient extends Configured implements MRConstants, Tool  {
    */
   public RunningJob getJob(JobID jobid) throws IOException {
     JobStatus status = jobSubmitClient.getJobStatus(jobid);
-    if (status != null) {
-      return new NetworkedJob(status);
+    JobProfile profile = jobSubmitClient.getJobProfile(jobid);
+    if (status != null && profile != null) {
+      return new NetworkedJob(status, profile, jobSubmitClient);
     } else {
       return null;
     }

src/test/org/apache/hadoop/mapred/TestNetworkedJob.java

@@ -0,0 +1,99 @@
+/**
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.*;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapreduce.Job;
+import org.junit.Test;
+import static org.mockito.Mockito.*;
+
+public class TestNetworkedJob {
+
+  @Test(expected=IOException.class)
+  public void testNullStatus() throws Exception {
+    JobProfile mockProf = mock(JobProfile.class);
+    JobSubmissionProtocol mockClient = mock(JobSubmissionProtocol.class);
+    new JobClient.NetworkedJob(null, mockProf, mockClient);
+  }
+  
+  @Test(expected=IOException.class)
+  public void testNullProfile() throws Exception {
+    JobStatus mockStatus = mock(JobStatus.class);
+    JobSubmissionProtocol mockClient = mock(JobSubmissionProtocol.class);
+    new JobClient.NetworkedJob(mockStatus, null, mockClient);
+  }
+
+  @Test(expected=IOException.class)
+  public void testNullClient() throws Exception {
+    JobStatus mockStatus = mock(JobStatus.class);
+    JobProfile mockProf = mock(JobProfile.class);
+    new JobClient.NetworkedJob(mockStatus, mockProf, null);
+  }
+  
+
+  @SuppressWarnings("deprecation")
+  @Test
+  public void testBadUpdate() throws Exception {
+    JobStatus mockStatus = mock(JobStatus.class);
+    JobProfile mockProf = mock(JobProfile.class);
+    JobSubmissionProtocol mockClient = mock(JobSubmissionProtocol.class);
+    
+    JobID id = new JobID("test",0);
+    
+    RunningJob rj = new JobClient.NetworkedJob(mockStatus, mockProf, mockClient);
+    
+    when(mockProf.getJobID()).thenReturn(id);
+    when(mockClient.getJobStatus(id)).thenReturn(null);
+    
+    boolean caught = false;
+    try {
+      rj.isSuccessful();
+    } catch(IOException e) {
+      caught = true;
+    }
+    assertTrue("Expected updateStatus to throw an IOException bt it did not", caught);
+    
+    //verification
+    verify(mockProf).getJobID();
+    verify(mockClient).getJobStatus(id);
+  }
+
+  @SuppressWarnings("deprecation")
+  @Test
+  public void testGetNullCounters() throws Exception {
+    JobStatus mockStatus = mock(JobStatus.class);
+    JobProfile mockProf = mock(JobProfile.class);
+    JobSubmissionProtocol mockClient = mock(JobSubmissionProtocol.class);
+    RunningJob underTest =
+      new JobClient.NetworkedJob(mockStatus, mockProf, mockClient); 
+
+    JobID id = new JobID("test", 0);
+    when(mockProf.getJobID()).thenReturn(id);
+    when(mockClient.getJobCounters(id)).thenReturn(null);
+    assertNull(underTest.getCounters());
+    //verification
+    verify(mockClient).getJobCounters(id);
+  }
+
+}