@Test
public void testNodeAffinityAssignment()
    throws InvalidAllocationTagsQueryException {

  PlacementConstraintManagerService pcm =
      new MemoryPlacementConstraintManager();
  AllocationTagsManager tm = new AllocationTagsManager(rmContext);
  // Register App1 with affinity constraint map
  pcm.registerApplication(appId1, constraintMap1);
  // No containers are running so all 'zk' and 'spark' allocations should fail
  // on every cluster NODE
  Iterator<RMNode> nodeIterator = rmNodes.iterator();
  while (nodeIterator.hasNext()) {
    RMNode currentNode = nodeIterator.next();
    SchedulerNode schedulerNode =newSchedulerNode(currentNode.getHostName(),
        currentNode.getRackName(), currentNode.getNodeID());
    Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
        createSchedulingRequest(sourceTag1), schedulerNode, pcm, tm));
    Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
        createSchedulingRequest(sourceTag2), schedulerNode, pcm, tm));
  }
  /**
   * Now place container:
   * Node0:123 (Rack1):
   *    container_app1_1 (hbase-m)
   */
  RMNode n0_r1 = rmNodes.get(0);
  RMNode n1_r1 = rmNodes.get(1);
  RMNode n2_r2 = rmNodes.get(2);
  RMNode n3_r2 = rmNodes.get(3);
  SchedulerNode schedulerNode0 =newSchedulerNode(n0_r1.getHostName(),
      n0_r1.getRackName(), n0_r1.getNodeID());
  SchedulerNode schedulerNode1 =newSchedulerNode(n1_r1.getHostName(),
      n1_r1.getRackName(), n1_r1.getNodeID());
  SchedulerNode schedulerNode2 =newSchedulerNode(n2_r2.getHostName(),
      n2_r2.getRackName(), n2_r2.getNodeID());
  SchedulerNode schedulerNode3 =newSchedulerNode(n3_r2.getHostName(),
      n3_r2.getRackName(), n3_r2.getNodeID());
  // 1 Containers on node 0 with allocationTag 'hbase-m'
  ContainerId hbase_m = ContainerId
      .newContainerId(ApplicationAttemptId.newInstance(appId1, 0), 0);
  tm.addContainer(n0_r1.getNodeID(), hbase_m, ImmutableSet.of("hbase-m"));
  // 'spark' placement on Node0 should now SUCCEED
  Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
      createSchedulingRequest(sourceTag1), schedulerNode0, pcm, tm));
  // FAIL on the rest of the nodes
  Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
      createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm));
  Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
      createSchedulingRequest(sourceTag1), schedulerNode2, pcm, tm));
  Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
      createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm));
  // Test diagnostics collector
  DiagnosticsCollector collector =
      new GenericDiagnosticsCollector();
  Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1,
      createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm,
      Optional.of(collector)));
  Assert.assertNotNull(collector.getDiagnostics());
  Assert.assertTrue(collector.getDiagnostics().contains("ALLOCATION_TAG"));
}