YARN-7774. Miscellaneous fixes to the PlacementProcessor. (asuresh)

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

@@ -469,6 +469,20 @@ public abstract class SchedulerNode {
     this.lastHeartbeatMonotonicTime = Time.monotonicNow();
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (!(o instanceof SchedulerNode)) return false;
+
+    SchedulerNode that = (SchedulerNode) o;
+
+    return getNodeID().equals(that.getNodeID());
+  }
+
+  @Override
+  public int hashCode() {
+    return getNodeID().hashCode();
+  }
 
   private static class ContainerInfo {
     private final RMContainer container;
@@ -479,4 +493,4 @@ public abstract class SchedulerNode {
       this.launchedOnNode = launchedOnNode;
     }
   }
-}
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java

@@ -2610,6 +2610,10 @@ public class CapacityScheduler extends
             " but only 1 will be attempted !!");
       }
       if (!appAttempt.isStopped()) {
+        Resource resource =
+            schedulingRequest.getResourceSizing().getResources();
+        schedulingRequest.getResourceSizing().setResources(
+            getNormalizedResource(resource));
         ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode>
             resourceCommitRequest = createResourceCommitRequest(
             appAttempt, schedulingRequest, schedulerNode);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java

@@ -99,14 +99,11 @@ public final class PlacementConstraintsUtil {
             targetApplicationId, te.getTargetValues(), Long::min);
       }
     }
-    // Make sure Anti-affinity satisfies hard upper limit
-    maxScopeCardinality = desiredMaxCardinality == 0 ? maxScopeCardinality - 1
-        : maxScopeCardinality;
 
     return (desiredMinCardinality <= 0
         || minScopeCardinality >= desiredMinCardinality) && (
         desiredMaxCardinality == Integer.MAX_VALUE
-            || maxScopeCardinality < desiredMaxCardinality);
+            || maxScopeCardinality <= desiredMaxCardinality);
   }
 
   private static boolean canSatisfyNodePartitionConstraintExpresssion(

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/algorithm/CircularIterator.java

@@ -0,0 +1,86 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.yarn.server.resourcemanager.scheduler.constraint.algorithm;
+
+import java.util.Iterator;
+
+/**
+ * Iterator that can take current state of an existing iterator
+ * and circularly iterate to that point.
+ */
+class CircularIterator<T> {
+  private Iterator<T> iterator = null;
+  private final Iterable<T> iterable;
+
+  private T startElem = null;
+  private T nextElem = null;
+
+  // if not null, This overrides the starting Element.
+  private T firstElem = null;
+
+  // Can't handle empty or null lists.
+  CircularIterator(T first, Iterator<T> iter,
+      Iterable<T> iterable) {
+    this.firstElem = first;
+    this.iterable = iterable;
+    if (!iter.hasNext()) {
+      this.iterator = this.iterable.iterator();
+    } else {
+      this.iterator = iter;
+    }
+    this.startElem = this.iterator.next();
+    this.nextElem = this.startElem;
+  }
+
+  boolean hasNext() {
+    if (this.nextElem != null || this.firstElem != null) {
+      return true;
+    } else {
+      if (this.iterator.hasNext()) {
+        T next = this.iterator.next();
+        if (this.startElem.equals(next)) {
+          return false;
+        } else {
+          this.nextElem = next;
+          return true;
+        }
+      } else {
+        this.iterator = this.iterable.iterator();
+        this.nextElem = this.iterator.next();
+        if (this.startElem.equals(this.nextElem)) {
+          return false;
+        }
+        return true;
+      }
+    }
+  }
+
+  T next() {
+    T retVal;
+    if (this.firstElem != null) {
+      retVal = this.firstElem;
+      this.firstElem = null;
+    } else if (this.nextElem != null) {
+      retVal = this.nextElem;
+      this.nextElem = null;
+    } else {
+      retVal = this.iterator.next();
+    }
+    return retVal;
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/algorithm/DefaultPlacementAlgorithm.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.algorithm;
 
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
@@ -49,6 +50,9 @@ public class DefaultPlacementAlgorithm implements ConstraintPlacementAlgorithm {
   private static final Logger LOG =
       LoggerFactory.getLogger(DefaultPlacementAlgorithm.class);
 
+  // Number of times to re-attempt placing a single scheduling request.
+  private static final int RE_ATTEMPT_COUNT = 2;
+
   private AllocationTagsManager tagsManager;
   private PlacementConstraintManager constraintManager;
   private NodeCandidateSelector nodeSelector;
@@ -85,16 +89,50 @@ public class DefaultPlacementAlgorithm implements ConstraintPlacementAlgorithm {
         new ConstraintPlacementAlgorithmOutput(requests.getApplicationId());
     List<SchedulerNode> allNodes = nodeSelector.selectNodes(null);
 
+    List<SchedulingRequest> rejectedRequests = new ArrayList<>();
+    int rePlacementCount = RE_ATTEMPT_COUNT;
+    while (rePlacementCount > 0) {
+      doPlacement(requests, resp, allNodes, rejectedRequests);
+      if (rejectedRequests.size() == 0 || rePlacementCount == 1) {
+        break;
+      }
+      requests = new BatchedRequests(requests.getIteratorType(),
+          requests.getApplicationId(), rejectedRequests,
+          requests.getPlacementAttempt());
+      rejectedRequests = new ArrayList<>();
+      rePlacementCount--;
+    }
+
+    resp.getRejectedRequests().addAll(rejectedRequests);
+    collector.collect(resp);
+    // Clean current temp-container tags
+    this.tagsManager.cleanTempContainers(requests.getApplicationId());
+  }
+
+  private void doPlacement(BatchedRequests requests,
+      ConstraintPlacementAlgorithmOutput resp,
+      List<SchedulerNode> allNodes,
+      List<SchedulingRequest> rejectedRequests) {
     Iterator<SchedulingRequest> requestIterator = requests.iterator();
+    Iterator<SchedulerNode> nIter = allNodes.iterator();
+    SchedulerNode lastSatisfiedNode = null;
     while (requestIterator.hasNext()) {
+      if (allNodes.isEmpty()) {
+        LOG.warn("No nodes available for placement at the moment !!");
+        break;
+      }
       SchedulingRequest schedulingRequest = requestIterator.next();
-      Iterator<SchedulerNode> nodeIter = allNodes.iterator();
+      CircularIterator<SchedulerNode> nodeIter =
+          new CircularIterator(lastSatisfiedNode, nIter, allNodes);
       int numAllocs = schedulingRequest.getResourceSizing().getNumAllocations();
       while (nodeIter.hasNext() && numAllocs > 0) {
         SchedulerNode node = nodeIter.next();
         try {
-          if (attemptPlacementOnNode(requests.getApplicationId(),
-              schedulingRequest, node)) {
+          String tag = schedulingRequest.getAllocationTags() == null ? "" :
+              schedulingRequest.getAllocationTags().iterator().next();
+          if (!requests.getBlacklist(tag).contains(node.getNodeID()) &&
+              attemptPlacementOnNode(
+                  requests.getApplicationId(), schedulingRequest, node)) {
             schedulingRequest.getResourceSizing()
                 .setNumAllocations(--numAllocs);
             PlacedSchedulingRequest placedReq =
@@ -108,6 +146,7 @@ public class DefaultPlacementAlgorithm implements ConstraintPlacementAlgorithm {
             this.tagsManager.addTempContainer(node.getNodeID(),
                 requests.getApplicationId(),
                 schedulingRequest.getAllocationTags());
+            lastSatisfiedNode = node;
           }
         } catch (InvalidAllocationTagsQueryException e) {
           LOG.warn("Got exception from TagManager !", e);
@@ -117,9 +156,6 @@ public class DefaultPlacementAlgorithm implements ConstraintPlacementAlgorithm {
     // Add all requests whose numAllocations still > 0 to rejected list.
     requests.getSchedulingRequests().stream()
         .filter(sReq -> sReq.getResourceSizing().getNumAllocations() > 0)
-        .forEach(rejReq -> resp.getRejectedRequests().add(rejReq));
-    collector.collect(resp);
-    // Clean current temp-container tags
-    this.tagsManager.cleanTempContainers(requests.getApplicationId());
+        .forEach(rejReq -> rejectedRequests.add(rejReq));
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/processor/BatchedRequests.java

@@ -133,4 +133,12 @@ public class BatchedRequests
   public Set<NodeId> getBlacklist(String tag) {
     return blacklist.getOrDefault(tag, Collections.EMPTY_SET);
   }
+
+  /**
+   * Get Iterator type.
+   * @return Iterator type.
+   */
+  public IteratorType getIteratorType() {
+    return iteratorType;
+  }
 }

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

@@ -263,7 +263,7 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
     }
 
     if (singleConstraint.getMinCardinality() != 0
-        || singleConstraint.getMaxCardinality() != 1) {
+        || singleConstraint.getMaxCardinality() != 0) {
       throwExceptionWithMetaInfo(
           "Only support anti-affinity, which is: minCardinality=0, "
               + "maxCardinality=1");

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java

@@ -311,7 +311,7 @@ public class MockAM {
             .allocationRequestId(allocationId).priority(priority)
             .allocationTags(allocationTags).placementConstraintExpression(
                 PlacementConstraints
-                    .targetCardinality(PlacementConstraints.NODE, 0, 1,
+                    .targetNotIn(PlacementConstraints.NODE,
                         PlacementConstraints.PlacementTargets
                             .allocationTagToIntraApp(targetTags)).build())
             .resourceSizing(resourceSizing).build()), null);
@@ -325,7 +325,7 @@ public class MockAM {
             ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
             .allocationRequestId(allocationId).priority(priority)
             .placementConstraintExpression(PlacementConstraints
-                .targetCardinality(PlacementConstraints.NODE, 0, 1,
+                .targetNotIn(PlacementConstraints.NODE,
                     PlacementConstraints.PlacementTargets
                         .allocationTagToIntraApp(tags),
                     PlacementConstraints.PlacementTargets

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementProcessor.java

@@ -153,13 +153,13 @@ public class TestPlacementProcessor {
   @Test(timeout = 300000)
   public void testCardinalityPlacement() throws Exception {
     HashMap<NodeId, MockNM> nodes = new HashMap<>();
-    MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm1 = new MockNM("h1:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm1.getNodeId(), nm1);
-    MockNM nm2 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm2 = new MockNM("h2:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm2.getNodeId(), nm2);
-    MockNM nm3 = new MockNM("h3:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm3 = new MockNM("h3:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm3.getNodeId(), nm3);
-    MockNM nm4 = new MockNM("h4:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm4 = new MockNM("h4:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm4.getNodeId(), nm4);
     nm1.registerNode();
     nm2.registerNode();
@@ -171,7 +171,7 @@ public class TestPlacementProcessor {
     MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2,
         Collections.singletonMap(Collections.singleton("foo"),
             PlacementConstraints.build(PlacementConstraints
-                .targetCardinality(NODE, 0, 4, allocationTag("foo")))));
+                .targetCardinality(NODE, 0, 3, allocationTag("foo")))));
     am1.addSchedulingRequest(
         Arrays.asList(schedulingRequest(1, 1, 1, 512, "foo"),
             schedulingRequest(1, 2, 1, 512, "foo"),
@@ -201,13 +201,13 @@ public class TestPlacementProcessor {
   @Test(timeout = 300000)
   public void testAffinityPlacement() throws Exception {
     HashMap<NodeId, MockNM> nodes = new HashMap<>();
-    MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm1 = new MockNM("h1:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm1.getNodeId(), nm1);
-    MockNM nm2 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm2 = new MockNM("h2:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm2.getNodeId(), nm2);
-    MockNM nm3 = new MockNM("h3:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm3 = new MockNM("h3:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm3.getNodeId(), nm3);
-    MockNM nm4 = new MockNM("h4:1234", 4096, rm.getResourceTrackerService());
+    MockNM nm4 = new MockNM("h4:1234", 8192, rm.getResourceTrackerService());
     nodes.put(nm4.getNodeId(), nm4);
     nm1.registerNode();
     nm2.registerNode();
@@ -267,7 +267,7 @@ public class TestPlacementProcessor {
         PlacementConstraints.build(targetIn(NODE, allocationTag("bar"))));
     // Containers with allocationTag 'foo' should not exceed 2 per NODE
     constraintMap.put(Collections.singleton("foo"), PlacementConstraints
-        .build(targetCardinality(NODE, 0, 2, allocationTag("foo"))));
+        .build(targetCardinality(NODE, 0, 1, allocationTag("foo"))));
     MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2, constraintMap);
     am1.addSchedulingRequest(
         Arrays.asList(schedulingRequest(1, 1, 1, 512, "bar"),
@@ -513,7 +513,8 @@ public class TestPlacementProcessor {
   private static void waitForContainerAllocation(Collection<MockNM> nodes,
       MockAM am, List<Container> allocatedContainers, int containerNum)
       throws Exception {
-    while (allocatedContainers.size() < containerNum) {
+    int attemptCount = 10;
+    while (allocatedContainers.size() < containerNum && attemptCount > 0) {
       for (MockNM node : nodes) {
         node.nodeHeartbeat(true);
       }
@@ -522,6 +523,7 @@ public class TestPlacementProcessor {
       sleep(1000);
       AllocateResponse allocResponse = am.schedule();
       allocatedContainers.addAll(allocResponse.getAllocatedContainers());
+      attemptCount--;
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/algorithm/TestCircularIterator.java

@@ -0,0 +1,84 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.yarn.server.resourcemanager.scheduler.constraint.algorithm;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Simple test case to test the Circular Iterator.
+ */
+public class TestCircularIterator {
+
+  @Test
+  public void testIteration() throws Exception {
+    List<String> list = Arrays.asList("a", "b", "c", "d");
+    CircularIterator<String> ci =
+        new CircularIterator<>(null, list.iterator(), list);
+    StringBuffer sb = new StringBuffer("");
+    while (ci.hasNext()) {
+      sb.append(ci.next());
+    }
+    Assert.assertEquals("abcd", sb.toString());
+
+    Iterator<String> lIter = list.iterator();
+    lIter.next();
+    lIter.next();
+    sb = new StringBuffer("");
+    ci = new CircularIterator<>(null, lIter, list);
+    while (ci.hasNext()) {
+      sb.append(ci.next());
+    }
+    Assert.assertEquals("cdab", sb.toString());
+
+    lIter = list.iterator();
+    lIter.next();
+    lIter.next();
+    lIter.next();
+    sb = new StringBuffer("");
+    ci = new CircularIterator<>("x", lIter, list);
+    while (ci.hasNext()) {
+      sb.append(ci.next());
+    }
+    Assert.assertEquals("xdabc", sb.toString());
+
+    list = Arrays.asList("a");
+    lIter = list.iterator();
+    lIter.next();
+    sb = new StringBuffer("");
+    ci = new CircularIterator<>("y", lIter, list);
+    while (ci.hasNext()) {
+      sb.append(ci.next());
+    }
+    Assert.assertEquals("ya", sb.toString());
+
+    try {
+      list = new ArrayList<>();
+      lIter = list.iterator();
+      new CircularIterator<>("y", lIter, list);
+      Assert.fail("Should fail..");
+    } catch (Exception e) {
+      // foo bar
+    }
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/TestSingleConstraintAppPlacementAllocator.java

@@ -118,7 +118,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer"),
                 PlacementConstraints.PlacementTargets.nodePartition(""))
@@ -134,7 +134,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer"),
                 PlacementConstraints.PlacementTargets.nodePartition("x"))
@@ -150,7 +150,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer")).build())
         .resourceSizing(
@@ -165,7 +165,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer")).build())
         .resourceSizing(
@@ -181,7 +181,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer")).build())
         .build(), true);
@@ -191,7 +191,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1).build())
+            .targetNotIn(PlacementConstraints.NODE).build())
         .build(), true);
 
     // Invalid (with multiple allocation tags expression specified)
@@ -199,7 +199,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper"),
                 PlacementConstraints.PlacementTargets
@@ -214,7 +214,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper"),
                 PlacementConstraints.PlacementTargets
@@ -255,7 +255,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.RACK, 0, 1,
+            .targetNotIn(PlacementConstraints.RACK,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer"),
                 PlacementConstraints.PlacementTargets.nodePartition(""))
@@ -268,7 +268,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer"),
                 PlacementConstraints.PlacementTargets.nodePartition(""))
@@ -284,7 +284,7 @@ public class TestSingleConstraintAppPlacementAllocator {
             ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
             .allocationRequestId(10L).priority(Priority.newInstance(1))
             .placementConstraintExpression(PlacementConstraints
-                .targetCardinality(PlacementConstraints.NODE, 0, 1,
+                .targetNotIn(PlacementConstraints.NODE,
                     PlacementConstraints.PlacementTargets
                         .allocationTagToIntraApp("mapper", "reducer"),
                     PlacementConstraints.PlacementTargets.nodePartition(""))
@@ -330,7 +330,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer"),
                 PlacementConstraints.PlacementTargets.nodePartition(""))
@@ -350,7 +350,7 @@ public class TestSingleConstraintAppPlacementAllocator {
             ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
             .allocationRequestId(10L).priority(Priority.newInstance(1))
             .placementConstraintExpression(PlacementConstraints
-                .targetCardinality(PlacementConstraints.NODE, 0, 1,
+                .targetNotIn(PlacementConstraints.NODE,
                     PlacementConstraints.PlacementTargets
                         .allocationTagToIntraApp("mapper", "reducer"),
                     PlacementConstraints.PlacementTargets.nodePartition(""))
@@ -372,7 +372,7 @@ public class TestSingleConstraintAppPlacementAllocator {
         ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED))
         .allocationRequestId(10L).priority(Priority.newInstance(1))
         .placementConstraintExpression(PlacementConstraints
-            .targetCardinality(PlacementConstraints.NODE, 0, 1,
+            .targetNotIn(PlacementConstraints.NODE,
                 PlacementConstraints.PlacementTargets
                     .allocationTagToIntraApp("mapper", "reducer"),
                 PlacementConstraints.PlacementTargets.nodePartition("x"))