MAPREDUCE-7151. RMContainerAllocator#handleJobPriorityChange expects application_priority always. Contributed by Bilwa S T.

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java

@@ -1020,12 +1020,14 @@ public class RMContainerAllocator extends RMContainerRequestor
     }
   }
 
-  private void handleJobPriorityChange(AllocateResponse response) {
-    Priority priorityFromResponse = Priority.newInstance(response
-        .getApplicationPriority().getPriority());
-
-    // Update the job priority to Job directly.
-    getJob().setJobPriority(priorityFromResponse);
+  void handleJobPriorityChange(AllocateResponse response) {
+    Priority applicationPriority = response.getApplicationPriority();
+    if (null != applicationPriority) {
+      Priority priorityFromResponse = Priority
+          .newInstance(applicationPriority.getPriority());
+      // Update the job priority to Job directly.
+      getJob().setJobPriority(priorityFromResponse);
+    }
   }
 
   @Private

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java

@@ -2160,6 +2160,20 @@ public class TestRMContainerAllocator {
     }
   }
 
+  @Test
+  public void testIfApplicationPriorityIsNotSet() {
+    Job mockJob = mock(Job.class);
+    RMCommunicator communicator = mock(RMCommunicator.class);
+    ClientService service = mock(ClientService.class);
+    AppContext context = mock(AppContext.class);
+    AMPreemptionPolicy policy = mock(AMPreemptionPolicy.class);
+    when(communicator.getJob()).thenReturn(mockJob);
+    RMContainerAllocator allocator = new RMContainerAllocator(service, context,
+        policy);
+    AllocateResponse response = Records.newRecord(AllocateResponse.class);
+    allocator.handleJobPriorityChange(response);
+  }
+
   @Test
   public void testReduceScheduling() throws Exception {
     int totalMaps = 10;