private static void validateResourceRequest(ResourceRequest resReq,
      Resource maximumAllocation, QueueInfo queueInfo, RMContext rmContext)
      throws InvalidResourceRequestException {
    final Resource requestedResource = resReq.getCapability();
    checkResourceRequestAgainstAvailableResource(requestedResource,
        maximumAllocation);

    String labelExp = resReq.getNodeLabelExpression();
    // we don't allow specify label expression other than resourceName=ANY now
    if (!ResourceRequest.ANY.equals(resReq.getResourceName())
        && labelExp != null && !labelExp.trim().isEmpty()) {
      throw new InvalidLabelResourceRequestException(
          "Invalid resource request, queue=" + queueInfo.getQueueName()
              + " specified node label expression in a "
              + "resource request has resource name = "
              + resReq.getResourceName());
    }

    // we don't allow specify label expression with more than one node labels now
    if (labelExp != null && labelExp.contains("&&")) {
      throw new InvalidLabelResourceRequestException(
          "Invalid resource request, queue=" + queueInfo.getQueueName()
              + " specified more than one node label "
              + "in a node label expression, node label expression = "
              + labelExp);
    }

    if (labelExp != null && !labelExp.trim().isEmpty() && queueInfo != null) {
      if (!checkQueueLabelExpression(queueInfo.getAccessibleNodeLabels(),
          labelExp, rmContext)) {
        throw new InvalidLabelResourceRequestException(
            "Invalid resource request" + ", queue=" + queueInfo.getQueueName()
                + " doesn't have permission to access all labels "
                + "in resource request. labelExpression of resource request="
                + labelExp + ". Queue labels="
                + (queueInfo.getAccessibleNodeLabels() == null ? ""
                    : StringUtils.join(
                        queueInfo.getAccessibleNodeLabels().iterator(), ',')));
      } else {
        checkQueueLabelInLabelManager(labelExp, rmContext);
      }
    }
  }
@Test(timeout = 30000)
  public void testValidateResourceRequest() throws IOException {
    ResourceScheduler mockScheduler = mock(ResourceScheduler.class);

    QueueInfo queueInfo = mock(QueueInfo.class);
    when(queueInfo.getQueueName()).thenReturn("queue");

    Resource maxResource =
            Resources.createResource(
                    YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
                    YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);

    when(rmContext.getScheduler()).thenReturn(mockScheduler);
    when(mockScheduler.getQueueInfo(Mockito.anyString(), Mockito.anyBoolean(),
        Mockito.anyBoolean())).thenReturn(queueInfo);

    // zero memory
    try {
      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, null,
              mockScheduler, rmContext, maxResource);
    } catch (InvalidResourceRequestException e) {
      fail("Zero memory should be accepted");
    }

    // zero vcores
    try {
      Resource resource =
              Resources.createResource(
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 0);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
    } catch (InvalidResourceRequestException e) {
      fail("Zero vcores should be accepted");
    }

    // max memory
    try {
      Resource resource =
              Resources.createResource(
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
    } catch (InvalidResourceRequestException e) {
      fail("Max memory should be accepted");
    }

    // max vcores
    try {
      Resource resource =
              Resources.createResource(
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
    } catch (InvalidResourceRequestException e) {
      fail("Max vcores should not be accepted");
    }

    // negative memory
    try {
      Resource resource =
              Resources.createResource(-1,
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
      fail("Negative memory should not be accepted");
    } catch (InvalidResourceRequestException e) {
      assertEquals(LESS_THAN_ZERO, e.getInvalidResourceType());
    }

    // negative vcores
    try {
      Resource resource =
              Resources.createResource(
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB, -1);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
      fail("Negative vcores should not be accepted");
    } catch (InvalidResourceRequestException e) {
      assertEquals(LESS_THAN_ZERO, e.getInvalidResourceType());
    }

    // more than max memory
    try {
      Resource resource =
              Resources.createResource(
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB + 1,
                      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
      fail("More than max memory should not be accepted");
    } catch (InvalidResourceRequestException e) {
      assertEquals(GREATER_THEN_MAX_ALLOCATION, e.getInvalidResourceType());
    }

    // more than max vcores
    try {
      Resource resource = Resources.createResource(
              YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
              YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES + 1);
      ResourceRequest resReq =
              BuilderUtils.newResourceRequest(mock(Priority.class),
                      ResourceRequest.ANY, resource, 1);
      normalizeAndvalidateRequest(resReq, null,
              mockScheduler, rmContext, maxResource);
      fail("More than max vcores should not be accepted");
    } catch (InvalidResourceRequestException e) {
      assertEquals(GREATER_THEN_MAX_ALLOCATION, e.getInvalidResourceType());
    }
  }