ResourceManager extends CompositeService implements Recoverable { @Private public ResourceScheduler getResourceScheduler() { return this.scheduler; } ResourceManager(); RMContext getRMContext(); static long getClusterTimeStamp(); void handleTransitionToStandBy(); @Private ClientRMService getClientRMService(); @Private ResourceScheduler getResourceScheduler(); @Private ResourceTrackerService getResourceTrackerService(); @Private ApplicationMasterService getApplicationMasterService(); @Private ApplicationACLsManager getApplicationACLsManager(); @Private QueueACLsManager getQueueACLsManager(); @Override void recover(RMState state); static void main(String argv[]); static InetSocketAddress getBindAddress(Configuration conf); static final int SHUTDOWN_HOOK_PRIORITY; }
@Test public void testResourceAllocation() throws IOException, YarnException, InterruptedException { LOG.info("--- START: testResourceAllocation ---"); final int memory = 4 * 1024; final int vcores = 4; String host1 = "host1"; org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm1 = registerNode(host1, 1234, 2345, NetworkTopology.DEFAULT_RACK, Resources.createResource(memory, vcores)); String host2 = "host2"; org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm2 = registerNode(host2, 1234, 2345, NetworkTopology.DEFAULT_RACK, Resources.createResource(memory/2, vcores/2)); Application application = new Application("user1", resourceManager); application.submit(); application.addNodeManager(host1, 1234, nm1); application.addNodeManager(host2, 1234, nm2); final int memory1 = 1024; Resource capability1 = Resources.createResource(memory1, 1); Priority priority1 = org.apache.hadoop.yarn.server.resourcemanager.resource.Priority.create(1); application.addResourceRequestSpec(priority1, capability1); Task t1 = new Task(application, priority1, new String[] {host1, host2}); application.addTask(t1); final int memory2 = 2048; Resource capability2 = Resources.createResource(memory2, 1); Priority priority0 = org.apache.hadoop.yarn.server.resourcemanager.resource.Priority.create(0); application.addResourceRequestSpec(priority0, capability2); application.schedule(); nodeUpdate(nm1); application.schedule(); checkResourceUsage(nm1, nm2); LOG.info("Adding new tasks..."); Task t2 = new Task(application, priority1, new String[] {host1, host2}); application.addTask(t2); Task t3 = new Task(application, priority0, new String[] {ResourceRequest.ANY}); application.addTask(t3); application.schedule(); checkResourceUsage(nm1, nm2); nodeUpdate(nm2); nodeUpdate(nm2); nodeUpdate(nm1); nodeUpdate(nm1); LOG.info("Trying to allocate..."); application.schedule(); checkResourceUsage(nm1, nm2); LOG.info("Finishing up tasks..."); application.finishTask(t1); application.finishTask(t2); application.finishTask(t3); AppAttemptRemovedSchedulerEvent appRemovedEvent1 = new AppAttemptRemovedSchedulerEvent( application.getApplicationAttemptId(), RMAppAttemptState.FINISHED, false); resourceManager.getResourceScheduler().handle(appRemovedEvent1); checkResourceUsage(nm1, nm2); LOG.info("--- END: testResourceAllocation ---"); }