@Test(timeout = 30000)
public void testValidateResourceRequestWithErrorLabelsPermission()
        throws IOException {

  // mock queue and scheduler
  ResourceScheduler scheduler = mock(ResourceScheduler.class);
  Set<String> queueAccessibleNodeLabels = Sets.newHashSet();
  QueueInfo queueInfo = mock(QueueInfo.class);
  when(queueInfo.getQueueName()).thenReturn("queue");
  when(queueInfo.getAccessibleNodeLabels())
          .thenReturn(queueAccessibleNodeLabels);
  when(scheduler.getQueueInfo(any(String.class), anyBoolean(), anyBoolean()))
          .thenReturn(queueInfo);
  when(rmContext.getScheduler()).thenReturn(scheduler);
  Resource maxResource = Resources.createResource(
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
  // queue has labels, success cases
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x"),
                    NodeLabel.newInstance("y")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression("y");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression("");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression(" ");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
  } catch (InvalidResourceRequestException e) {
    e.printStackTrace();
    fail("Should be valid when request labels is a subset of queue labels");
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x", "y"));
  }
  // same as above, but cluster node labels don't contains label being
  // requested. should fail
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  }
  // queue has labels, failed cases (when ask a label not included by queue)
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x"),
                    NodeLabel.newInstance("y")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("z");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x", "y"));
  }
  // we don't allow specify more than two node labels in a single expression
  // now
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x"),
                    NodeLabel.newInstance("y")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x && y");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x", "y"));
  }
  // queue doesn't have label, succeed (when request no label)
  queueAccessibleNodeLabels.clear();
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression("");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression("  ");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
  } catch (InvalidResourceRequestException e) {
    e.printStackTrace();
    fail("Should be valid when request labels is empty");
  }
  boolean invalidlabelexception = false;
  // queue doesn't have label, failed (when request any label)
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidLabelResourceRequestException e) {
    invalidlabelexception = true;
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x"));
  }
  Assert.assertTrue("InvalidLabelResourceRequestException expected",
          invalidlabelexception);
  // queue is "*", always succeeded
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.add(RMNodeLabelsManager.ANY);
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x"),
                    NodeLabel.newInstance("y"), NodeLabel.newInstance("z")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression("y");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    resReq.setNodeLabelExpression("z");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
  } catch (InvalidResourceRequestException e) {
    e.printStackTrace();
    fail("Should be valid when queue can access any labels");
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x", "y", "z"));
  }
  // same as above, but cluster node labels don't contains label, should fail
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.add(RMNodeLabelsManager.ANY);
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  }
  // we don't allow resource name other than ANY and specify label
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x"),
                    NodeLabel.newInstance("y")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), "rack", resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x", "y"));
  }
  // we don't allow resource name other than ANY and specify label even if
  // queue has accessible label = *
  try {
    // set queue accessible node labesl to *
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays
            .asList(CommonNodeLabelsManager.ANY));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
            ImmutableSet.of(NodeLabel.newInstance("x")));
    Resource resource = Resources.createResource(
            0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
            mock(Priority.class), "rack", resource, 1);
    resReq.setNodeLabelExpression("x");
    normalizeAndvalidateRequest(resReq, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
            Arrays.asList("x"));
  }
  try {
    Resource resource = Resources.createResource(0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq1 = BuilderUtils
            .newResourceRequest(mock(Priority.class), "*", resource, 1, "x");
    normalizeAndvalidateRequest(resReq1, "queue",
            scheduler, rmContext, maxResource);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
    assertEquals("Invalid label resource request, cluster do not contain , "
            + "label= x", e.getMessage());
  }
  try {
    rmContext.getYarnConfiguration()
            .set(YarnConfiguration.NODE_LABELS_ENABLED, "false");
    Resource resource = Resources.createResource(0,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq1 = BuilderUtils
            .newResourceRequest(mock(Priority.class), "*", resource, 1, "x");
    normalizeAndvalidateRequest(resReq1, "queue",
            scheduler, rmContext, maxResource);
    Assert.assertEquals(RMNodeLabelsManager.NO_LABEL,
            resReq1.getNodeLabelExpression());
  } catch (InvalidResourceRequestException e) {
    assertEquals("Invalid resource request, node label not enabled but "
            + "request contains label expression", e.getMessage());
  }
}