RMServerUtils { public static ContainerUpdates validateAndSplitUpdateResourceRequests(RMContext rmContext, AllocateRequest request, Resource maximumAllocation, List<UpdateContainerError> updateErrors) { ContainerUpdates updateRequests = new ContainerUpdates(); Set<ContainerId> outstandingUpdate = new HashSet<>(); for (UpdateContainerRequest updateReq : request.getUpdateRequests()) { RMContainer rmContainer = rmContext.getScheduler().getRMContainer( updateReq.getContainerId()); String msg = validateContainerIdAndVersion(outstandingUpdate, updateReq, rmContainer); ContainerUpdateType updateType = updateReq.getContainerUpdateType(); if (msg == null) { if ((updateType != ContainerUpdateType.PROMOTE_EXECUTION_TYPE) && (updateType !=ContainerUpdateType.DEMOTE_EXECUTION_TYPE)) { if (validateIncreaseDecreaseRequest( rmContext, updateReq, maximumAllocation)) { if (ContainerUpdateType.INCREASE_RESOURCE == updateType) { updateRequests.getIncreaseRequests().add(updateReq); } else { updateRequests.getDecreaseRequests().add(updateReq); } outstandingUpdate.add(updateReq.getContainerId()); } else { msg = RESOURCE_OUTSIDE_ALLOWED_RANGE; } } else { ExecutionType original = rmContainer.getExecutionType(); ExecutionType target = updateReq.getExecutionType(); if (target != original) { if (target == ExecutionType.GUARANTEED && original == ExecutionType.OPPORTUNISTIC) { updateRequests.getPromotionRequests().add(updateReq); outstandingUpdate.add(updateReq.getContainerId()); } else if (target == ExecutionType.OPPORTUNISTIC && original == ExecutionType.GUARANTEED) { updateRequests.getDemotionRequests().add(updateReq); outstandingUpdate.add(updateReq.getContainerId()); } } } } checkAndcreateUpdateError(updateErrors, updateReq, rmContainer, msg); } return updateRequests; }  static List<RMNode> queryRMNodes(RMContext context,
      EnumSet<NodeState> acceptedStates); static ContainerUpdates validateAndSplitUpdateResourceRequests(RMContext rmContext,
      AllocateRequest request, Resource maximumAllocation,
      List<UpdateContainerError> updateErrors); static void normalizeAndValidateRequests(List<ResourceRequest> ask,
      Resource maximumAllocation, String queueName, YarnScheduler scheduler,
      RMContext rmContext); static void checkSchedContainerChangeRequest(
      SchedContainerChangeRequest request, boolean increase); static void validateBlacklistRequest(
      ResourceBlacklistRequest blacklistRequest); static void validateContainerReleaseRequest(List<ContainerId> containerReleaseList,
      ApplicationAttemptId appAttemptId); static UserGroupInformation verifyAdminAccess(
      YarnAuthorizationProvider authorizer, String method, final Log LOG); static UserGroupInformation verifyAdminAccess(
      YarnAuthorizationProvider authorizer, String method, String module,
      final Log LOG); static YarnApplicationState createApplicationState(
      RMAppState rmAppState); static YarnApplicationAttemptState createApplicationAttemptState(
      RMAppAttemptState rmAppAttemptState); static void processRMProxyUsersConf(Configuration conf); static void validateApplicationTimeouts(
      Map<ApplicationTimeoutType, Long> timeouts); static Map<ApplicationTimeoutType, Long> validateISO8601AndConvertToLocalTimeEpoch(
      Map<ApplicationTimeoutType, String> timeoutsInISO8601); static int getApplicableNodeCountForAM(RMContext rmContext,
      Configuration conf, List<ResourceRequest> amReqs); static Long getOrDefault(Map<String, Long> map, String key,
      Long defaultValue); static final String UPDATE_OUTSTANDING_ERROR; static final String RESOURCE_OUTSIDE_ALLOWED_RANGE; static final ApplicationResourceUsageReport DUMMY_APPLICATION_RESOURCE_USAGE_REPORT; }
@Test public void testValidateAndSplitUpdateResourceRequests() { List<UpdateContainerRequest> updateRequests = new ArrayList<>(); int containerVersion = 10; int resource = 10; Resource maxAllocation = Resource.newInstance(resource, resource); UpdateContainerRequestPBImpl updateContainerRequestPBFail = new UpdateContainerRequestPBImpl(); updateContainerRequestPBFail.setContainerVersion(containerVersion); updateContainerRequestPBFail .setCapability(Resource.newInstance(resource + 1, resource + 1)); updateContainerRequestPBFail .setContainerId(Mockito.mock(ContainerId.class)); ContainerId containerIdOk = Mockito.mock(ContainerId.class); Resource capabilityOk = Resource.newInstance(resource - 1, resource - 1); UpdateContainerRequestPBImpl updateContainerRequestPBOk = new UpdateContainerRequestPBImpl(); updateContainerRequestPBOk.setContainerVersion(containerVersion); updateContainerRequestPBOk.setCapability(capabilityOk); updateContainerRequestPBOk.setContainerUpdateType(INCREASE_RESOURCE); updateContainerRequestPBOk.setContainerId(containerIdOk); updateRequests.add(updateContainerRequestPBOk); updateRequests.add(updateContainerRequestPBFail); Dispatcher dispatcher = Mockito.mock(Dispatcher.class); RMContext rmContext = Mockito.mock(RMContext.class); ResourceScheduler scheduler = Mockito.mock(ResourceScheduler.class); Mockito.when(rmContext.getScheduler()).thenReturn(scheduler); Mockito.when(rmContext.getDispatcher()).thenReturn(dispatcher); RMContainer rmContainer = Mockito.mock(RMContainer.class); Mockito.when(scheduler.getRMContainer(Mockito.any())) .thenReturn(rmContainer); Container container = Mockito.mock(Container.class); Mockito.when(container.getVersion()).thenReturn(containerVersion); Mockito.when(rmContainer.getContainer()).thenReturn(container); Mockito.when(scheduler.getNormalizedResource(capabilityOk, maxAllocation)) .thenReturn(capabilityOk); AllocateRequest allocateRequest = AllocateRequest.newInstance(1, 0.5f, new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>(), updateRequests, null); List<UpdateContainerError> updateErrors = new ArrayList<>(); ContainerUpdates containerUpdates = RMServerUtils.validateAndSplitUpdateResourceRequests(rmContext, allocateRequest, maxAllocation, updateErrors); Assert.assertEquals(1, updateErrors.size()); Assert.assertEquals(resource + 1, updateErrors.get(0) .getUpdateContainerRequest().getCapability().getMemorySize()); Assert.assertEquals(resource + 1, updateErrors.get(0) .getUpdateContainerRequest().getCapability().getVirtualCores()); Assert.assertEquals(RESOURCE_OUTSIDE_ALLOWED_RANGE, updateErrors.get(0).getReason()); Assert.assertEquals(1, containerUpdates.getIncreaseRequests().size()); UpdateContainerRequest increaseRequest = containerUpdates.getIncreaseRequests().get(0); Assert.assertEquals(capabilityOk.getVirtualCores(), increaseRequest.getCapability().getVirtualCores()); Assert.assertEquals(capabilityOk.getMemorySize(), increaseRequest.getCapability().getMemorySize()); Assert.assertEquals(containerIdOk, increaseRequest.getContainerId()); }