@Test (timeout = 30000) public void testNormalizeNodeLabelExpression() throws IOException { YarnScheduler scheduler = mock(YarnScheduler.class); Set<String> queueAccessibleNodeLabels = Sets.newHashSet(); QueueInfo queueInfo = mock(QueueInfo.class); when(queueInfo.getQueueName()).thenReturn("queue"); when(queueInfo.getAccessibleNodeLabels()).thenReturn(queueAccessibleNodeLabels); when(queueInfo.getDefaultNodeLabelExpression()).thenReturn(" x "); when(scheduler.getQueueInfo(any(String.class), anyBoolean(), anyBoolean())) .thenReturn(queueInfo); Resource maxResource = Resources.createResource( YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES, YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_GCORES); try { queueAccessibleNodeLabels.clear(); queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y")); rmContext.getNodeLabelManager().addToCluserNodeLabels( ImmutableSet.of("x", "y")); Resource resource = Resources.createResource( 0, YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES, YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_GCORES); ResourceRequest resReq = BuilderUtils.newResourceRequest( mock(Priority.class), ResourceRequest.ANY, resource, 1); SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue", scheduler, rmContext); Assert.assertTrue(resReq.getNodeLabelExpression().equals("x")); resReq.setNodeLabelExpression(" y "); SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue", scheduler, rmContext); Assert.assertTrue(resReq.getNodeLabelExpression().equals("y")); } 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")); } }
public static void normalizeAndValidateRequest(ResourceRequest resReq, Resource maximumResource, String queueName, YarnScheduler scheduler, boolean isRecovery, RMContext rmContext, QueueInfo queueInfo) throws InvalidResourceRequestException { if (queueInfo == null) { try { queueInfo = scheduler.getQueueInfo(queueName, false, false); } catch (IOException e) { } } SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo); if (!isRecovery) { validateResourceRequest(resReq, maximumResource, queueInfo, rmContext); } }
SchedulerUtils { public static void normalizeAndValidateRequest(ResourceRequest resReq, Resource maximumResource, String queueName, YarnScheduler scheduler, boolean isRecovery, RMContext rmContext, QueueInfo queueInfo) throws InvalidResourceRequestException { if (queueInfo == null) { try { queueInfo = scheduler.getQueueInfo(queueName, false, false); } catch (IOException e) { } } SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo); if (!isRecovery) { validateResourceRequest(resReq, maximumResource, queueInfo, rmContext); } } }
SchedulerUtils { public static void normalizeAndValidateRequest(ResourceRequest resReq, Resource maximumResource, String queueName, YarnScheduler scheduler, boolean isRecovery, RMContext rmContext, QueueInfo queueInfo) throws InvalidResourceRequestException { if (queueInfo == null) { try { queueInfo = scheduler.getQueueInfo(queueName, false, false); } catch (IOException e) { } } SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo); if (!isRecovery) { validateResourceRequest(resReq, maximumResource, queueInfo, rmContext); } }  }
SchedulerUtils { public static void normalizeAndValidateRequest(ResourceRequest resReq, Resource maximumResource, String queueName, YarnScheduler scheduler, boolean isRecovery, RMContext rmContext, QueueInfo queueInfo) throws InvalidResourceRequestException { if (queueInfo == null) { try { queueInfo = scheduler.getQueueInfo(queueName, false, false); } catch (IOException e) { } } SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo); if (!isRecovery) { validateResourceRequest(resReq, maximumResource, queueInfo, rmContext); } }  static ContainerStatus createAbnormalContainerStatus(
      ContainerId containerId, String diagnostics); static ContainerStatus createPreemptedContainerStatus(
      ContainerId containerId, String diagnostics); static void normalizeRequests(
    List<ResourceRequest> asks,
    ResourceCalculator resourceCalculator,
    Resource clusterResource,
    Resource minimumResource,
    Resource maximumResource); static void normalizeRequest(
    ResourceRequest ask,
    ResourceCalculator resourceCalculator,
    Resource clusterResource,
    Resource minimumResource,
    Resource maximumResource); static void normalizeRequests(
      List<ResourceRequest> asks,
      ResourceCalculator resourceCalculator, 
      Resource clusterResource,
      Resource minimumResource,
      Resource maximumResource,
      Resource incrementResource); static void normalizeRequest(
      ResourceRequest ask, 
      ResourceCalculator resourceCalculator, 
      Resource clusterResource,
      Resource minimumResource,
      Resource maximumResource,
      Resource incrementResource); static void normalizeAndValidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      boolean isRecovery, RMContext rmContext, QueueInfo queueInfo); static void normalizeAndValidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      boolean isRecovery, RMContext rmContext); static void normalizeAndvalidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      RMContext rmContext, QueueInfo queueInfo); static void normalizeAndvalidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      RMContext rmContext); static boolean checkQueueAccessToNode(Set<String> queueLabels,
      Set<String> nodeLabels); static void checkIfLabelInClusterNodeLabels(RMNodeLabelsManager mgr,
      Set<String> labels); static boolean checkNodeLabelExpression(Set<String> nodeLabels,
      String labelExpression); static boolean checkQueueLabelExpression(Set<String> queueLabels,
      String labelExpression, RMContext rmContext); static AccessType toAccessType(QueueACL acl); }
SchedulerUtils { public static void normalizeAndValidateRequest(ResourceRequest resReq, Resource maximumResource, String queueName, YarnScheduler scheduler, boolean isRecovery, RMContext rmContext, QueueInfo queueInfo) throws InvalidResourceRequestException { if (queueInfo == null) { try { queueInfo = scheduler.getQueueInfo(queueName, false, false); } catch (IOException e) { } } SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo); if (!isRecovery) { validateResourceRequest(resReq, maximumResource, queueInfo, rmContext); } }  static ContainerStatus createAbnormalContainerStatus(
      ContainerId containerId, String diagnostics); static ContainerStatus createPreemptedContainerStatus(
      ContainerId containerId, String diagnostics); static void normalizeRequests(
    List<ResourceRequest> asks,
    ResourceCalculator resourceCalculator,
    Resource clusterResource,
    Resource minimumResource,
    Resource maximumResource); static void normalizeRequest(
    ResourceRequest ask,
    ResourceCalculator resourceCalculator,
    Resource clusterResource,
    Resource minimumResource,
    Resource maximumResource); static void normalizeRequests(
      List<ResourceRequest> asks,
      ResourceCalculator resourceCalculator, 
      Resource clusterResource,
      Resource minimumResource,
      Resource maximumResource,
      Resource incrementResource); static void normalizeRequest(
      ResourceRequest ask, 
      ResourceCalculator resourceCalculator, 
      Resource clusterResource,
      Resource minimumResource,
      Resource maximumResource,
      Resource incrementResource); static void normalizeAndValidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      boolean isRecovery, RMContext rmContext, QueueInfo queueInfo); static void normalizeAndValidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      boolean isRecovery, RMContext rmContext); static void normalizeAndvalidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      RMContext rmContext, QueueInfo queueInfo); static void normalizeAndvalidateRequest(ResourceRequest resReq,
      Resource maximumResource, String queueName, YarnScheduler scheduler,
      RMContext rmContext); static boolean checkQueueAccessToNode(Set<String> queueLabels,
      Set<String> nodeLabels); static void checkIfLabelInClusterNodeLabels(RMNodeLabelsManager mgr,
      Set<String> labels); static boolean checkNodeLabelExpression(Set<String> nodeLabels,
      String labelExpression); static boolean checkQueueLabelExpression(Set<String> queueLabels,
      String labelExpression, RMContext rmContext); static AccessType toAccessType(QueueACL acl); static final String RELEASED_CONTAINER; static final String LOST_CONTAINER; static final String PREEMPTED_CONTAINER; static final String COMPLETED_APPLICATION; static final String EXPIRED_CONTAINER; static final String UNRESERVED_CONTAINER; }