@Test public void testMaxAllocationAfterUpdateNodeResource() throws IOException { final int configuredMaxVCores = 20; final int configuredMaxMemory = 10 * 1024; Resource configuredMaximumResource = Resource.newInstance (configuredMaxMemory, configuredMaxVCores, configuredMaxVCores); configureScheduler(); 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, 5); Resource resource2 = Resource.newInstance(4096, 10, 10); Resource resource3 = Resource.newInstance(512, 1, 1); Resource resource4 = Resource.newInstance(1024, 2, 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(); } }
public synchronized void updateNodeResource(RMNode nm, ResourceOption resourceOption) { SchedulerNode node = getSchedulerNode(nm.getNodeID()); Resource newResource = resourceOption.getResource(); Resource oldResource = node.getTotalResource(); if(!oldResource.equals(newResource)) { rmContext.getNodeLabelManager().updateNodeResource(nm.getNodeID(), newResource); LOG.info("Update resource on node: " + node.getNodeName() + " from: " + oldResource + ", to: " + newResource); nodes.remove(nm.getNodeID()); updateMaximumAllocation(node, false); node.setTotalResource(newResource); nodes.put(nm.getNodeID(), (N)node); updateMaximumAllocation(node, true); Resources.subtractFrom(clusterResource, oldResource); Resources.addTo(clusterResource, newResource); } else { LOG.warn("Update resource on node: " + node.getNodeName() + " with the same resource: " + newResource); } }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { public synchronized void updateNodeResource(RMNode nm, ResourceOption resourceOption) { SchedulerNode node = getSchedulerNode(nm.getNodeID()); Resource newResource = resourceOption.getResource(); Resource oldResource = node.getTotalResource(); if(!oldResource.equals(newResource)) { rmContext.getNodeLabelManager().updateNodeResource(nm.getNodeID(), newResource); LOG.info("Update resource on node: " + node.getNodeName() + " from: " + oldResource + ", to: " + newResource); nodes.remove(nm.getNodeID()); updateMaximumAllocation(node, false); node.setTotalResource(newResource); nodes.put(nm.getNodeID(), (N)node); updateMaximumAllocation(node, true); Resources.subtractFrom(clusterResource, oldResource); Resources.addTo(clusterResource, newResource); } else { LOG.warn("Update resource on node: " + node.getNodeName() + " with the same resource: " + newResource); } } }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { public synchronized void updateNodeResource(RMNode nm, ResourceOption resourceOption) { SchedulerNode node = getSchedulerNode(nm.getNodeID()); Resource newResource = resourceOption.getResource(); Resource oldResource = node.getTotalResource(); if(!oldResource.equals(newResource)) { rmContext.getNodeLabelManager().updateNodeResource(nm.getNodeID(), newResource); LOG.info("Update resource on node: " + node.getNodeName() + " from: " + oldResource + ", to: " + newResource); nodes.remove(nm.getNodeID()); updateMaximumAllocation(node, false); node.setTotalResource(newResource); nodes.put(nm.getNodeID(), (N)node); updateMaximumAllocation(node, true); Resources.subtractFrom(clusterResource, oldResource); Resources.addTo(clusterResource, newResource); } else { LOG.warn("Update resource on node: " + node.getNodeName() + " with the same resource: " + newResource); } } AbstractYarnScheduler(String name); }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { public synchronized void updateNodeResource(RMNode nm, ResourceOption resourceOption) { SchedulerNode node = getSchedulerNode(nm.getNodeID()); Resource newResource = resourceOption.getResource(); Resource oldResource = node.getTotalResource(); if(!oldResource.equals(newResource)) { rmContext.getNodeLabelManager().updateNodeResource(nm.getNodeID(), newResource); LOG.info("Update resource on node: " + node.getNodeName() + " from: " + oldResource + ", to: " + newResource); nodes.remove(nm.getNodeID()); updateMaximumAllocation(node, false); node.setTotalResource(newResource); nodes.put(nm.getNodeID(), (N)node); updateMaximumAllocation(node, true); Resources.subtractFrom(clusterResource, oldResource); Resources.addTo(clusterResource, newResource); } else { LOG.warn("Update resource on node: " + node.getNodeName() + " with the same resource: " + newResource); } } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); synchronized List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); 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); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm); SchedulerNode getSchedulerNode(NodeId nodeId); @Override synchronized void moveAllApps(String sourceQueue, String destQueue); @Override synchronized void killAllAppsInQueue(String queueName); synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId); }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { public synchronized void updateNodeResource(RMNode nm, ResourceOption resourceOption) { SchedulerNode node = getSchedulerNode(nm.getNodeID()); Resource newResource = resourceOption.getResource(); Resource oldResource = node.getTotalResource(); if(!oldResource.equals(newResource)) { rmContext.getNodeLabelManager().updateNodeResource(nm.getNodeID(), newResource); LOG.info("Update resource on node: " + node.getNodeName() + " from: " + oldResource + ", to: " + newResource); nodes.remove(nm.getNodeID()); updateMaximumAllocation(node, false); node.setTotalResource(newResource); nodes.put(nm.getNodeID(), (N)node); updateMaximumAllocation(node, true); Resources.subtractFrom(clusterResource, oldResource); Resources.addTo(clusterResource, newResource); } else { LOG.warn("Update resource on node: " + node.getNodeName() + " with the same resource: " + newResource); } } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); synchronized List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); 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); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm); SchedulerNode getSchedulerNode(NodeId nodeId); @Override synchronized void moveAllApps(String sourceQueue, String destQueue); @Override synchronized void killAllAppsInQueue(String queueName); synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId);  }