FifoScheduler extends
    AbstractYarnScheduler<FifoAppAttempt, FiCaSchedulerNode> implements
    Configurable { public Resource getUsedResource() { return usedResource; } FifoScheduler(); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void setConf(Configuration conf); @Override synchronized Configuration getConf(); @Override int getNumClusterNodes(); @Override synchronized void setRMContext(RMContext rmContext); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @Override Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests,
      List<ContainerId> release, List<String> blacklistAdditions, List<String> blacklistRemovals,
      ContainerUpdates updateRequests); @VisibleForTesting synchronized void addApplication(ApplicationId applicationId,
      String queue, String user, boolean isAppRecovering); @VisibleForTesting synchronized void addApplicationAttempt(ApplicationAttemptId appAttemptId,
          boolean transferStateFromPreviousAttempt,
          boolean isAttemptRecovering); @Override void handle(SchedulerEvent event); @Override QueueInfo getQueueInfo(String queueName,
      boolean includeChildQueues, boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override ResourceCalculator getResourceCalculator(); @Override void recover(RMState state); @Override RMContainer getRMContainer(ContainerId containerId); @Override QueueMetrics getRootQueueMetrics(); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override synchronized List<ApplicationAttemptId> getAppsInQueue(String queueName); Resource getUsedResource(); @VisibleForTesting @Override void killContainer(RMContainer container); @Override synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm);  }
@Test(timeout = 60000) public void testFifoScheduling() throws Exception { GenericTestUtils.setRootLogLevel(Level.DEBUG); MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); MockNM nm2 = rm.registerNode("127.0.0.2:5678", 4 * GB); RMApp app1 = MockRMAppSubmitter.submitWithMemory(2048, rm); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); SchedulerNodeReport report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(2 * GB, report_nm1.getUsedResource().getMemorySize()); RMApp app2 = MockRMAppSubmitter.submitWithMemory(2048, rm); nm2.nodeHeartbeat(true); RMAppAttempt attempt2 = app2.getCurrentAppAttempt(); MockAM am2 = rm.sendAMLaunched(attempt2.getAppAttemptId()); am2.registerAppAttempt(); SchedulerNodeReport report_nm2 = rm.getResourceScheduler().getNodeReport(nm2.getNodeId()); Assert.assertEquals(2 * GB, report_nm2.getUsedResource().getMemorySize()); am1.addRequests(new String[] { "127.0.0.1", "127.0.0.2" }, GB, 1, 1); AllocateResponse alloc1Response = am1.schedule(); am2.addRequests(new String[] { "127.0.0.1", "127.0.0.2" }, 3 * GB, 0, 1); AllocateResponse alloc2Response = am2.schedule(); nm1.nodeHeartbeat(true); while (alloc1Response.getAllocatedContainers().size() < 1) { LOG.info("Waiting for containers to be created for app 1..."); Thread.sleep(1000); alloc1Response = am1.schedule(); } while (alloc2Response.getAllocatedContainers().size() < 1) { LOG.info("Waiting for containers to be created for app 2..."); Thread.sleep(1000); alloc2Response = am2.schedule(); } nm2.nodeHeartbeat(true); List<Container> allocated1 = alloc1Response.getAllocatedContainers(); Assert.assertEquals(1, allocated1.size()); Assert.assertEquals(1 * GB, allocated1.get(0).getResource().getMemorySize()); Assert.assertEquals(nm1.getNodeId(), allocated1.get(0).getNodeId()); List<Container> allocated2 = alloc2Response.getAllocatedContainers(); Assert.assertEquals(1, allocated2.size()); Assert.assertEquals(3 * GB, allocated2.get(0).getResource().getMemorySize()); Assert.assertEquals(nm1.getNodeId(), allocated2.get(0).getNodeId()); report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); report_nm2 = rm.getResourceScheduler().getNodeReport(nm2.getNodeId()); Assert.assertEquals(0, report_nm1.getAvailableResource().getMemorySize()); Assert.assertEquals(2 * GB, report_nm2.getAvailableResource().getMemorySize()); Assert.assertEquals(6 * GB, report_nm1.getUsedResource().getMemorySize()); Assert.assertEquals(2 * GB, report_nm2.getUsedResource().getMemorySize()); Container c1 = allocated1.get(0); Assert.assertEquals(GB, c1.getResource().getMemorySize()); ContainerStatus containerStatus = BuilderUtils.newContainerStatus(c1.getId(), ContainerState.COMPLETE, "", 0, c1.getResource()); nm1.containerStatus(containerStatus); int waitCount = 0; while (attempt1.getJustFinishedContainers().size() < 1 && waitCount++ != 20) { LOG.info("Waiting for containers to be finished for app 1... Tried " + waitCount + " times already.."); Thread.sleep(1000); } Assert.assertEquals(1, attempt1.getJustFinishedContainers().size()); Assert.assertEquals(1, am1.schedule().getCompletedContainersStatuses() .size()); report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(5 * GB, report_nm1.getUsedResource().getMemorySize()); rm.stop(); }
@Test(timeout = 60000) public void testResourceOverCommit() throws Exception { int waitCount; MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 4 * GB); RMApp app1 = MockRMAppSubmitter.submitWithMemory(2048, rm); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); SchedulerNodeReport report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(2 * GB, report_nm1.getUsedResource().getMemorySize()); Assert.assertEquals(2 * GB, report_nm1.getAvailableResource().getMemorySize()); am1.addRequests(new String[] { "127.0.0.1", "127.0.0.2" }, 2 * GB, 1, 1); AllocateResponse alloc1Response = am1.schedule(); nm1.nodeHeartbeat(true); while (alloc1Response.getAllocatedContainers().size() < 1) { LOG.info("Waiting for containers to be created for app 1..."); Thread.sleep(1000); alloc1Response = am1.schedule(); } List<Container> allocated1 = alloc1Response.getAllocatedContainers(); Assert.assertEquals(1, allocated1.size()); Assert.assertEquals(2 * GB, allocated1.get(0).getResource().getMemorySize()); Assert.assertEquals(nm1.getNodeId(), allocated1.get(0).getNodeId()); report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(0, report_nm1.getAvailableResource().getMemorySize()); Assert.assertEquals(4 * GB, report_nm1.getUsedResource().getMemorySize()); Container c1 = allocated1.get(0); Assert.assertEquals(2 * GB, c1.getResource().getMemorySize()); Map<NodeId, ResourceOption> nodeResourceMap = new HashMap<NodeId, ResourceOption>(); nodeResourceMap.put(nm1.getNodeId(), ResourceOption.newInstance(Resource.newInstance(2 * GB, 1), -1)); UpdateNodeResourceRequest request = UpdateNodeResourceRequest.newInstance(nodeResourceMap); rm.getAdminService().updateNodeResource(request); waitCount = 0; while (waitCount++ != 20) { report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); if (null != report_nm1 && report_nm1.getAvailableResource().getMemorySize() != 0) { break; } LOG.info("Waiting for RMNodeResourceUpdateEvent to be handled... Tried " + waitCount + " times already.."); Thread.sleep(1000); } report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(4 * GB, report_nm1.getUsedResource().getMemorySize()); Assert.assertEquals(-2 * GB, report_nm1.getAvailableResource().getMemorySize()); ContainerStatus containerStatus = BuilderUtils.newContainerStatus(c1.getId(), ContainerState.COMPLETE, "", 0, c1.getResource()); nm1.containerStatus(containerStatus); waitCount = 0; while (attempt1.getJustFinishedContainers().size() < 1 && waitCount++ != 20) { LOG.info("Waiting for containers to be finished for app 1... Tried " + waitCount + " times already.."); Thread.sleep(100); } Assert.assertEquals(1, attempt1.getJustFinishedContainers().size()); Assert.assertEquals(1, am1.schedule().getCompletedContainersStatuses() .size()); report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(2 * GB, report_nm1.getUsedResource().getMemorySize()); Assert.assertEquals(0 * GB, report_nm1.getAvailableResource().getMemorySize()); rm.stop(); }