YARN-8367. Fix NPE in SingleConstraintAppPlacementAllocator when placement constraint in SchedulingRequest is null. Contributed by Weiwei Yang.

(Cherry picked from commit 6468071f13)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
 import org.apache.commons.collections.IteratorUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -238,14 +239,21 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
           "Only GUARANTEED execution type is supported.");
     }
 
+    // Node partition
+    String nodePartition = null;
+    // Target allocation tags
+    Set<String> targetAllocationTags = null;
+
     PlacementConstraint constraint =
         newSchedulingRequest.getPlacementConstraint();
 
+    if (constraint != null) {
       // We only accept SingleConstraint
-    PlacementConstraint.AbstractConstraint ac = constraint.getConstraintExpr();
+      PlacementConstraint.AbstractConstraint ac = constraint
+          .getConstraintExpr();
       if (!(ac instanceof PlacementConstraint.SingleConstraint)) {
-      throwExceptionWithMetaInfo(
+        throwExceptionWithMetaInfo("Only accepts "
-          "Only accepts " + PlacementConstraint.SingleConstraint.class.getName()
+            + PlacementConstraint.SingleConstraint.class.getName()
                 + " as constraint-expression. Rejecting the new added "
             + "constraint-expression.class=" + ac.getClass().getName());
       }
@@ -277,13 +285,8 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
             "TargetExpression should not be null or empty");
       }
 
-    // Set node partition
+      for (PlacementConstraint.TargetExpression targetExpression :
-    String nodePartition = null;
+          targetExpressionSet) {
-
-    // Target allocation tags
-    Set<String> targetAllocationTags = null;
-
-    for (PlacementConstraint.TargetExpression targetExpression : targetExpressionSet) {
         // Handle node partition
         if (targetExpression.getTargetType().equals(
             PlacementConstraint.TargetExpression.TargetType.NODE_ATTRIBUTE)) {
@@ -296,8 +299,8 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
           }
 
           if (nodePartition != null) {
-          // This means we have duplicated node partition entry inside placement
+            // This means we have duplicated node partition entry
-          // constraint, which might be set by mistake.
+            // inside placement constraint, which might be set by mistake.
             throwExceptionWithMetaInfo(
                 "Only one node partition targetExpression is allowed");
           }
@@ -324,8 +327,8 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
                 "Only one AllocationTag targetExpression is allowed");
           }
 
-        if (targetExpression.getTargetValues() == null || targetExpression
+          if (targetExpression.getTargetValues() == null ||
-            .getTargetValues().isEmpty()) {
+              targetExpression.getTargetValues().isEmpty()) {
             throwExceptionWithMetaInfo("Failed to find allocation tags from "
                 + "TargetExpressions or couldn't find self-app target.");
           }
@@ -338,10 +341,16 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
       if (targetAllocationTags == null) {
         // That means we don't have ALLOCATION_TAG specified
         throwExceptionWithMetaInfo(
-          "Couldn't find target expression with type == ALLOCATION_TAG, it is "
+            "Couldn't find target expression with type == ALLOCATION_TAG,"
-              + "required to include one and only one target expression with "
+                + " it is required to include one and only one target"
-              + "type == ALLOCATION_TAG");
+                + " expression with type == ALLOCATION_TAG");
+      }
+    }
 
+    // If this scheduling request doesn't contain a placement constraint,
+    // we set allocation tags an empty set.
+    if (targetAllocationTags == null) {
+      targetAllocationTags = ImmutableSet.of();
     }
 
     if (nodePartition == null) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java

@@ -18,8 +18,15 @@
 
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
+import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
+import org.apache.hadoop.yarn.api.records.SchedulingRequest;
+import org.apache.hadoop.yarn.api.resource.PlacementConstraint;
+import org.apache.hadoop.yarn.api.resource.PlacementConstraints;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.TargetApplicationsNamespace;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.Resource;
@@ -39,6 +46,8 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.apache.hadoop.yarn.api.resource.PlacementConstraints.PlacementTargets.*;
+
 public class TestSchedulingRequestContainerAllocation {
   private final int GB = 1024;
 
@@ -393,4 +402,79 @@ public class TestSchedulingRequestContainerAllocation {
     Assert.assertTrue(caughtException);
     rm1.close();
   }
+
+  @Test
+  public void testSchedulingRequestWithNullConstraint() throws Exception {
+    Configuration csConf = TestUtils.getConfigurationWithMultipleQueues(
+        new Configuration());
+    csConf.set(YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_HANDLER,
+        YarnConfiguration.SCHEDULER_RM_PLACEMENT_CONSTRAINTS_HANDLER);
+
+    // inject node label manager
+    MockRM rm1 = new MockRM(csConf) {
+      @Override
+      public RMNodeLabelsManager createNodeLabelManager() {
+        return mgr;
+      }
+    };
+
+    rm1.getRMContext().setNodeLabelManager(mgr);
+    rm1.start();
+
+    // 4 NMs.
+    MockNM[] nms = new MockNM[4];
+    RMNode[] rmNodes = new RMNode[4];
+    for (int i = 0; i < 4; i++) {
+      nms[i] = rm1.registerNode("192.168.0." + i + ":1234", 10 * GB);
+      rmNodes[i] = rm1.getRMContext().getRMNodes().get(nms[i].getNodeId());
+    }
+
+    // app1 -> c
+    RMApp app1 = rm1.submitApp(1 * GB, "app", "user", null, "c");
+    MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nms[0]);
+
+    CapacityScheduler cs = (CapacityScheduler) rm1.getResourceScheduler();
+
+    PlacementConstraint constraint = PlacementConstraints
+        .targetNotIn("node", allocationTag("t1"))
+        .build();
+    SchedulingRequest sc = SchedulingRequest
+        .newInstance(0, Priority.newInstance(1),
+            ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED),
+            ImmutableSet.of("t1"),
+            ResourceSizing.newInstance(1, Resource.newInstance(1024, 1)),
+            constraint);
+    AllocateRequest request = AllocateRequest.newBuilder()
+        .schedulingRequests(ImmutableList.of(sc)).build();
+    am1.allocate(request);
+
+    for (int i = 0; i < 4; i++) {
+      cs.handle(new NodeUpdateSchedulerEvent(rmNodes[i]));
+    }
+
+    FiCaSchedulerApp schedApp = cs.getApplicationAttempt(
+        am1.getApplicationAttemptId());
+    Assert.assertEquals(2, schedApp.getLiveContainers().size());
+
+
+    // Send another request with null placement constraint,
+    // ensure there is no NPE while handling this request.
+    sc = SchedulingRequest
+        .newInstance(1, Priority.newInstance(1),
+            ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED),
+            ImmutableSet.of("t2"),
+            ResourceSizing.newInstance(2, Resource.newInstance(1024, 1)),
+            null);
+    AllocateRequest request1 = AllocateRequest.newBuilder()
+        .schedulingRequests(ImmutableList.of(sc)).build();
+    am1.allocate(request1);
+
+    for (int i = 0; i < 4; i++) {
+      cs.handle(new NodeUpdateSchedulerEvent(rmNodes[i]));
+    }
+
+    Assert.assertEquals(4, schedApp.getLiveContainers().size());
+
+    rm1.close();
+  }
 }