AbstractYarnScheduler extends AbstractService implements ResourceScheduler { public void updateNodeResource(RMNode nm, ResourceOption resourceOption) { writeLock.lock(); try { SchedulerNode node = getSchedulerNode(nm.getNodeID()); Resource newResource = resourceOption.getResource(); final int timeout = resourceOption.getOverCommitTimeout(); Resource oldResource = node.getTotalResource(); if (!oldResource.equals(newResource)) { rmContext.getNodeLabelManager().updateNodeResource(nm.getNodeID(), newResource); LOG.info("Update resource on node: {} from: {}, to: {} in {} ms", node.getNodeName(), oldResource, newResource, timeout); nodeTracker.removeNode(nm.getNodeID()); node.updateTotalResource(newResource); node.setOvercommitTimeOut(timeout); signalContainersIfOvercommitted(node, timeout == 0); nodeTracker.addNode((N) node); } else{ LOG.warn("Update resource on node: " + node.getNodeName() + " with the same resource: " + newResource); } } finally { writeLock.unlock(); } } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); @VisibleForTesting ClusterNodeTracker<N> getNodeTracker(); @VisibleForTesting SchedulingMonitorManager getSchedulingMonitorManager(); List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); List<N> getBlacklistedNodes(final SchedulerApplicationAttempt app); List<N> getNodes(final NodeFilter filter); boolean shouldContainersBeAutoUpdated(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); SchedulerHealth getSchedulerHealth(); long getLastNodeUpdateTime(); T getApplicationAttempt(ApplicationAttemptId applicationAttemptId); @Override SchedulerAppReport getSchedulerAppInfo(
      ApplicationAttemptId appAttemptId); @Override ApplicationResourceUsageReport getAppResourceUsageReport(
      ApplicationAttemptId appAttemptId); T getCurrentAttemptForContainer(ContainerId containerId); @Override RMContainer getRMContainer(ContainerId containerId); @Override SchedulerNodeReport getNodeReport(NodeId nodeId); @Override String moveApplication(ApplicationId appId, String newQueue); @Override void preValidateMoveApplication(ApplicationId appId,
      String newQueue); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); void recoverContainersOnNode(List<NMContainerStatus> containerReports,
      RMNode nm); @VisibleForTesting void clearPendingContainerCache(); @VisibleForTesting @Private // clean up a completed container void completedContainer(RMContainer rmContainer,
      ContainerStatus containerStatus, RMContainerEventType event); @Override N getSchedulerNode(NodeId nodeId); @Override void moveAllApps(String sourceQueue, String destQueue); @Override void killAllAppsInQueue(String queueName); void updateNodeResource(RMNode nm,
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); boolean placementConstraintEnabled(); @Override List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId); @Override List<SchedulingRequest> getPendingSchedulingRequestsForAttempt(
      ApplicationAttemptId attemptId); @Override Priority checkAndGetApplicationPriority(
          Priority priorityRequestedByApp, UserGroupInformation user,
          String queuePath, ApplicationId applicationId); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); @Override Priority getMaxClusterLevelAppPriority(); @Override void setClusterMaxPriority(Configuration conf); ActivitiesManager getActivitiesManager(); Clock getClock(); @VisibleForTesting void setClock(Clock clock); @Lock(Lock.NoLock.class) SchedulerNode getNode(NodeId nodeId); @Override Resource getNormalizedResource(Resource requestedResource,
                                        Resource maxResourceCapability); @Override List<NodeId> getNodeIds(String resourceName); void asyncContainerRelease(RMContainer container); Resource getMinimumAllocation(); Resource getMaximumAllocation(); @Override long checkAndGetApplicationLifetime(String queueName, long lifetime); @Override long getMaximumApplicationLifetime(String queueName); @VisibleForTesting abstract void killContainer(RMContainer container); @VisibleForTesting void update(); @Override void reinitialize(Configuration conf, RMContext rmContext); @Override boolean attemptAllocationOnNode(SchedulerApplicationAttempt appAttempt,
      SchedulingRequest schedulingRequest, SchedulerNode schedulerNode); @Override void resetSchedulerMetrics();  }
@Test public void testMaxAllocationAfterUpdateNodeResource() throws IOException { final int configuredMaxVCores = 20; final int configuredMaxMemory = 10 * 1024; Resource configuredMaximumResource = Resource.newInstance (configuredMaxMemory, configuredMaxVCores); YarnConfiguration conf = getConf(); conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES, configuredMaxVCores); conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, configuredMaxMemory); conf.setLong( YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_SCHEDULING_WAIT_MS, 0); MockRM rm = new MockRM(conf); try { rm.start(); AbstractYarnScheduler scheduler = (AbstractYarnScheduler) rm .getResourceScheduler(); verifyMaximumResourceCapability(configuredMaximumResource, scheduler); Resource resource1 = Resource.newInstance(2048, 5); Resource resource2 = Resource.newInstance(4096, 10); Resource resource3 = Resource.newInstance(512, 1); Resource resource4 = Resource.newInstance(1024, 2); RMNode node1 = MockNodes.newNodeInfo( 0, resource1, 1, "127.0.0.2"); scheduler.handle(new NodeAddedSchedulerEvent(node1)); RMNode node2 = MockNodes.newNodeInfo( 0, resource3, 2, "127.0.0.3"); scheduler.handle(new NodeAddedSchedulerEvent(node2)); verifyMaximumResourceCapability(resource1, scheduler); scheduler.updateNodeResource(node1, ResourceOption.newInstance( resource2, 0)); verifyMaximumResourceCapability(resource2, scheduler); scheduler.updateNodeResource(node1, ResourceOption.newInstance( resource1, 0)); verifyMaximumResourceCapability(resource1, scheduler); scheduler.updateNodeResource(node2, ResourceOption.newInstance( resource4, 0)); verifyMaximumResourceCapability(resource1, scheduler); scheduler.updateNodeResource(node2, ResourceOption.newInstance( resource3, 0)); verifyMaximumResourceCapability(resource1, scheduler); } finally { rm.stop(); } }