OpportunisticContainerAllocatorAMService extends ApplicationMasterService implements DistributedSchedulingAMProtocol,
    EventHandler<SchedulerEvent> { @VisibleForTesting synchronized List<RemoteNode> getLeastLoadedNodes() { long currTime = System.currentTimeMillis(); if ((currTime - lastCacheUpdateTime > cacheRefreshInterval) || (cachedNodes == null)) { cachedNodes = convertToRemoteNodes( this.nodeMonitor.selectLeastLoadedNodes(this.numNodes)); if (cachedNodes.size() > 0) { lastCacheUpdateTime = currTime; } } return cachedNodes; } OpportunisticContainerAllocatorAMService(RMContext rmContext,
      YarnScheduler scheduler); @Override Server getServer(YarnRPC rpc, Configuration serverConf,
      InetSocketAddress addr, AMRMTokenSecretManager secretManager); @Override RegisterDistributedSchedulingAMResponse registerApplicationMasterForDistributedScheduling(
      RegisterApplicationMasterRequest request); @Override DistributedSchedulingAllocateResponse allocateForDistributedScheduling(
      DistributedSchedulingAllocateRequest request); @Override void handle(SchedulerEvent event);  }
@Test(timeout = 600000) public void testContainerPromoteAndDemoteBeforeContainerStart() throws Exception { HashMap<NodeId, MockNM> nodes = new HashMap<>(); MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm1.getNodeId(), nm1); MockNM nm2 = new MockNM("h1:4321", 4096, rm.getResourceTrackerService()); nodes.put(nm2.getNodeId(), nm2); MockNM nm3 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm3.getNodeId(), nm3); MockNM nm4 = new MockNM("h2:4321", 4096, rm.getResourceTrackerService()); nodes.put(nm4.getNodeId(), nm4); nm1.registerNode(); nm2.registerNode(); nm3.registerNode(); nm4.registerNode(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); nm3.nodeHeartbeat(oppContainersStatus, true); nm4.nodeHeartbeat(oppContainersStatus, true); OpportunisticContainerAllocatorAMService amservice = (OpportunisticContainerAllocatorAMService) rm .getApplicationMasterService(); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm) .withAppName("app") .withUser("user") .withAcls(null) .withQueue("default") .withUnmanagedAM(false) .build(); RMApp app1 = MockRMAppSubmitter.submit(rm, data); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2); ResourceScheduler scheduler = rm.getResourceScheduler(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); nm3.nodeHeartbeat(oppContainersStatus, true); nm4.nodeHeartbeat(oppContainersStatus, true); GenericTestUtils.waitFor(() -> amservice.getLeastLoadedNodes().size() == 4, 10, 10 * 100); QueueMetrics metrics = ((CapacityScheduler) scheduler).getRootQueue() .getMetrics(); verifyMetrics(metrics, 15360, 15, 1024, 1, 1); AllocateResponse allocateResponse = am1.allocate( Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2, true, null, ExecutionTypeRequest.newInstance( ExecutionType.OPPORTUNISTIC, true))), null); List<Container> allocatedContainers = allocateResponse .getAllocatedContainers(); Assert.assertEquals(2, allocatedContainers.size()); Container container = allocatedContainers.get(0); MockNM allocNode = nodes.get(container.getNodeId()); MockNM sameHostDiffNode = null; for (NodeId n : nodes.keySet()) { if (n.getHost().equals(allocNode.getNodeId().getHost()) && n.getPort() != allocNode.getNodeId().getPort()) { sameHostDiffNode = nodes.get(n); } } verifyMetrics(metrics, 15360, 15, 1024, 1, 1); am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(0, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); sameHostDiffNode.nodeHeartbeat(oppContainersStatus, true); rm.drainEvents(); allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>()); Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size()); dispatcher.waitForEventThreadToWait(); rm.drainEvents(); verifyMetrics(metrics, 15360, 15, 1024, 1, 1); allocateResponse = am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(0, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size()); Assert.assertEquals(1, allocateResponse.getUpdateErrors().size()); Assert.assertEquals("UPDATE_OUTSTANDING_ERROR", allocateResponse.getUpdateErrors().get(0).getReason()); Assert.assertEquals(container.getId(), allocateResponse.getUpdateErrors().get(0) .getUpdateContainerRequest().getContainerId()); allocateResponse = am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(1, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size()); Assert.assertEquals(1, allocateResponse.getUpdateErrors().size()); Assert.assertEquals("INCORRECT_CONTAINER_VERSION_ERROR", allocateResponse.getUpdateErrors().get(0).getReason()); Assert.assertEquals(0, allocateResponse.getUpdateErrors().get(0) .getCurrentContainerVersion()); Assert.assertEquals(container.getId(), allocateResponse.getUpdateErrors().get(0) .getUpdateContainerRequest().getContainerId()); allocNode.nodeHeartbeat(oppContainersStatus, true); rm.drainEvents(); allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>()); Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); Container uc = allocateResponse.getUpdatedContainers().get(0).getContainer(); Assert.assertEquals(ExecutionType.GUARANTEED, uc.getExecutionType()); Assert.assertEquals(uc.getId(), container.getId()); Assert.assertEquals(uc.getVersion(), container.getVersion() + 1); verifyMetrics(metrics, 14336, 14, 2048, 2, 2); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); nm3.nodeHeartbeat(oppContainersStatus, true); nm4.nodeHeartbeat(oppContainersStatus, true); rm.drainEvents(); RMContainer rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( uc.getId().getApplicationAttemptId()).getRMContainer(uc.getId()); Assert.assertEquals(RMContainerState.ACQUIRED, rmContainer.getState()); allocateResponse = am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(uc.getVersion(), uc.getId(), ContainerUpdateType.DEMOTE_EXECUTION_TYPE, null, ExecutionType.OPPORTUNISTIC))); Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); uc = allocateResponse.getUpdatedContainers().get(0).getContainer(); Assert.assertEquals(ExecutionType.OPPORTUNISTIC, uc.getExecutionType()); Assert.assertEquals(uc.getId(), container.getId()); Assert.assertEquals(uc.getVersion(), container.getVersion() + 2); dispatcher.waitForEventThreadToWait(); rm.drainEvents(); verifyMetrics(metrics, 15360, 15, 1024, 1, 1); }
@Test(timeout = 60000) public void testContainerPromoteAfterContainerStart() throws Exception { HashMap<NodeId, MockNM> nodes = new HashMap<>(); MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm1.getNodeId(), nm1); MockNM nm2 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm2.getNodeId(), nm2); nm1.registerNode(); nm2.registerNode(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); OpportunisticContainerAllocatorAMService amservice = (OpportunisticContainerAllocatorAMService) rm .getApplicationMasterService(); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm) .withAppName("app") .withUser("user") .withAcls(null) .withQueue("default") .withUnmanagedAM(false) .build(); RMApp app1 = MockRMAppSubmitter.submit(rm, data); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2); ResourceScheduler scheduler = rm.getResourceScheduler(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); GenericTestUtils.waitFor(() -> amservice.getLeastLoadedNodes().size() == 2, 10, 10 * 100); QueueMetrics metrics = ((CapacityScheduler) scheduler).getRootQueue() .getMetrics(); verifyMetrics(metrics, 7168, 7, 1024, 1, 1); AllocateResponse allocateResponse = am1.allocate( Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2, true, null, ExecutionTypeRequest.newInstance( ExecutionType.OPPORTUNISTIC, true))), null); List<Container> allocatedContainers = allocateResponse .getAllocatedContainers(); Assert.assertEquals(2, allocatedContainers.size()); Container container = allocatedContainers.get(0); MockNM allocNode = nodes.get(container.getNodeId()); allocNode.nodeHeartbeat(Arrays.asList( ContainerStatus.newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); RMContainer rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( container.getId().getApplicationAttemptId()).getRMContainer( container.getId()); Assert.assertEquals(RMContainerState.RUNNING, rmContainer.getState()); verifyMetrics(metrics, 7168, 7, 1024, 1, 1); am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(0, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); verifyMetrics(metrics, 7168, 7, 1024, 1, 1); allocateResponse = am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(0, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size()); Assert.assertEquals(1, allocateResponse.getUpdateErrors().size()); Assert.assertEquals("UPDATE_OUTSTANDING_ERROR", allocateResponse.getUpdateErrors().get(0).getReason()); Assert.assertEquals(container.getId(), allocateResponse.getUpdateErrors().get(0) .getUpdateContainerRequest().getContainerId()); allocNode.nodeHeartbeat(Arrays.asList( ContainerStatus.newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>()); Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); Container uc = allocateResponse.getUpdatedContainers().get(0).getContainer(); Assert.assertEquals(ExecutionType.GUARANTEED, uc.getExecutionType()); Assert.assertEquals(uc.getId(), container.getId()); Assert.assertEquals(uc.getVersion(), container.getVersion() + 1); rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( uc.getId().getApplicationAttemptId()).getRMContainer(uc.getId()); Assert.assertEquals(RMContainerState.RUNNING, rmContainer.getState()); verifyMetrics(metrics, 6144, 6, 2048, 2, 2); }
@Test(timeout = 600000) public void testContainerPromoteAfterContainerComplete() throws Exception { HashMap<NodeId, MockNM> nodes = new HashMap<>(); MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm1.getNodeId(), nm1); MockNM nm2 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm2.getNodeId(), nm2); nm1.registerNode(); nm2.registerNode(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); OpportunisticContainerAllocatorAMService amservice = (OpportunisticContainerAllocatorAMService) rm .getApplicationMasterService(); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm) .withAppName("app") .withUser("user") .withAcls(null) .withQueue("default") .withUnmanagedAM(false) .build(); RMApp app1 = MockRMAppSubmitter.submit(rm, data); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2); ResourceScheduler scheduler = rm.getResourceScheduler(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); GenericTestUtils.waitFor(() -> amservice.getLeastLoadedNodes().size() == 2, 10, 10 * 100); QueueMetrics metrics = ((CapacityScheduler) scheduler).getRootQueue() .getMetrics(); verifyMetrics(metrics, 7168, 7, 1024, 1, 1); AllocateResponse allocateResponse = am1.allocate( Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2, true, null, ExecutionTypeRequest.newInstance( ExecutionType.OPPORTUNISTIC, true))), null); List<Container> allocatedContainers = allocateResponse .getAllocatedContainers(); Assert.assertEquals(2, allocatedContainers.size()); Container container = allocatedContainers.get(0); MockNM allocNode = nodes.get(container.getNodeId()); allocNode.nodeHeartbeat(Arrays.asList( ContainerStatus.newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); RMContainer rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( container.getId().getApplicationAttemptId()).getRMContainer( container.getId()); Assert.assertEquals(RMContainerState.RUNNING, rmContainer.getState()); allocNode.nodeHeartbeat(Arrays.asList( ContainerStatus.newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.COMPLETE, "", 0)), true); rm.drainEvents(); rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( container.getId().getApplicationAttemptId()).getRMContainer( container.getId()); Assert.assertNull(rmContainer); verifyMetrics(metrics, 7168, 7, 1024, 1, 1); allocateResponse = am1.sendContainerUpdateRequest( Arrays.asList(UpdateContainerRequest.newInstance(0, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); Assert.assertEquals(1, allocateResponse.getCompletedContainersStatuses().size()); Assert.assertEquals(container.getId(), allocateResponse.getCompletedContainersStatuses().get(0) .getContainerId()); Assert.assertEquals(0, allocateResponse.getUpdatedContainers().size()); Assert.assertEquals(1, allocateResponse.getUpdateErrors().size()); Assert.assertEquals("INVALID_CONTAINER_ID", allocateResponse.getUpdateErrors().get(0).getReason()); Assert.assertEquals(container.getId(), allocateResponse.getUpdateErrors().get(0) .getUpdateContainerRequest().getContainerId()); verifyMetrics(metrics, 7168, 7, 1024, 1, 1); }
@Test(timeout = 600000) public void testContainerAutoUpdateContainer() throws Exception { rm.stop(); createAndStartRMWithAutoUpdateContainer(); MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService()); nm1.registerNode(); nm1.nodeHeartbeat(oppContainersStatus, true); OpportunisticContainerAllocatorAMService amservice = (OpportunisticContainerAllocatorAMService) rm .getApplicationMasterService(); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm) .withAppName("app") .withUser("user") .withAcls(null) .withQueue("default") .withUnmanagedAM(false) .build(); RMApp app1 = MockRMAppSubmitter.submit(rm, data); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm1); ResourceScheduler scheduler = rm.getResourceScheduler(); RMNode rmNode1 = rm.getRMContext().getRMNodes().get(nm1.getNodeId()); nm1.nodeHeartbeat(oppContainersStatus, true); GenericTestUtils.waitFor(() -> amservice.getLeastLoadedNodes().size() == 1, 10, 10 * 100); AllocateResponse allocateResponse = am1.allocate(Arrays.asList( ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2, true, null, ExecutionTypeRequest .newInstance(ExecutionType.OPPORTUNISTIC, true))), null); List<Container> allocatedContainers = allocateResponse.getAllocatedContainers(); allocatedContainers.addAll( am1.allocate(null, null).getAllocatedContainers()); Assert.assertEquals(2, allocatedContainers.size()); Container container = allocatedContainers.get(0); nm1.nodeHeartbeat(Arrays.asList(ContainerStatus .newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); RMContainer rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt(container.getId().getApplicationAttemptId()) .getRMContainer(container.getId()); Assert.assertEquals(RMContainerState.RUNNING, rmContainer.getState()); allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList( UpdateContainerRequest.newInstance(0, container.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE, null, ExecutionType.GUARANTEED))); nm1.nodeHeartbeat(Arrays.asList(ContainerStatus .newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>()); Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); UpdatedContainer uc = allocateResponse.getUpdatedContainers().get(0); Assert.assertEquals(container.getId(), uc.getContainer().getId()); Assert.assertEquals(ExecutionType.GUARANTEED, uc.getContainer().getExecutionType()); NodeHeartbeatResponse response = nm1.nodeHeartbeat(true); Assert.assertEquals(1, response.getContainersToUpdate().size()); Container containersFromNM = response.getContainersToUpdate().get(0); Assert.assertEquals(container.getId(), containersFromNM.getId()); Assert.assertEquals(ExecutionType.GUARANTEED, containersFromNM.getExecutionType()); allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList( UpdateContainerRequest.newInstance(1, container.getId(), ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(2 * GB, 1), null))); response = nm1.nodeHeartbeat(Arrays.asList(ContainerStatus .newInstance(container.getId(), ExecutionType.GUARANTEED, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); if (allocateResponse.getUpdatedContainers().size() == 0) { allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>()); } Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); uc = allocateResponse.getUpdatedContainers().get(0); Assert.assertEquals(container.getId(), uc.getContainer().getId()); Assert.assertEquals(Resource.newInstance(2 * GB, 1), uc.getContainer().getResource()); rm.drainEvents(); if (response.getContainersToUpdate().size() == 0) { response = nm1.nodeHeartbeat(true); } Assert.assertEquals(1, response.getContainersToUpdate().size()); Assert.assertEquals(Resource.newInstance(2 * GB, 1), response.getContainersToUpdate().get(0).getResource()); allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList( UpdateContainerRequest.newInstance(2, container.getId(), ContainerUpdateType.DECREASE_RESOURCE, Resources.createResource(1 * GB, 1), null))); Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); rm.drainEvents(); response = nm1.nodeHeartbeat(true); Assert.assertEquals(1, response.getContainersToUpdate().size()); Assert.assertEquals(Resource.newInstance(1 * GB, 1), response.getContainersToUpdate().get(0).getResource()); nm1.nodeHeartbeat(oppContainersStatus, true); allocateResponse = am1.sendContainerUpdateRequest(Arrays.asList( UpdateContainerRequest.newInstance(3, container.getId(), ContainerUpdateType.DEMOTE_EXECUTION_TYPE, null, ExecutionType.OPPORTUNISTIC))); response = nm1.nodeHeartbeat(Arrays.asList(ContainerStatus .newInstance(container.getId(), ExecutionType.GUARANTEED, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); if (allocateResponse.getUpdatedContainers().size() == 0) { allocateResponse = am1.allocate(new ArrayList<>(), new ArrayList<>()); } Assert.assertEquals(1, allocateResponse.getUpdatedContainers().size()); uc = allocateResponse.getUpdatedContainers().get(0); Assert.assertEquals(ExecutionType.OPPORTUNISTIC, uc.getContainer().getExecutionType()); if (response.getContainersToUpdate().size() == 0) { response = nm1.nodeHeartbeat(oppContainersStatus, true); } Assert.assertEquals(1, response.getContainersToUpdate().size()); Assert.assertEquals(ExecutionType.OPPORTUNISTIC, response.getContainersToUpdate().get(0).getExecutionType()); }
@Test(timeout = 60000) public void testOpportunisticSchedulerMetrics() throws Exception { HashMap<NodeId, MockNM> nodes = new HashMap<>(); MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm1.getNodeId(), nm1); MockNM nm2 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService()); nodes.put(nm2.getNodeId(), nm2); nm1.registerNode(); nm2.registerNode(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); OpportunisticSchedulerMetrics metrics = OpportunisticSchedulerMetrics.getMetrics(); int allocContainers = metrics.getAllocatedContainers(); long aggrAllocatedContainers = metrics.getAggregatedAllocatedContainers(); long aggrOffSwitchContainers = metrics.getAggregatedOffSwitchContainers(); long aggrReleasedContainers = metrics.getAggregatedReleasedContainers(); OpportunisticContainerAllocatorAMService amservice = (OpportunisticContainerAllocatorAMService) rm .getApplicationMasterService(); RMApp app1 = MockRMAppSubmitter.submit(rm, MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm) .withAppName("app") .withUser("user") .withAcls(null) .withQueue("default") .build()); ApplicationAttemptId attemptId = app1.getCurrentAppAttempt().getAppAttemptId(); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2); ResourceScheduler scheduler = rm.getResourceScheduler(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); GenericTestUtils.waitFor(() -> amservice.getLeastLoadedNodes().size() == 2, 10, 10 * 100); AllocateResponse allocateResponse = am1.allocate(Arrays.asList( ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2, true, null, ExecutionTypeRequest .newInstance(ExecutionType.OPPORTUNISTIC, true))), null); List<Container> allocatedContainers = allocateResponse .getAllocatedContainers(); Assert.assertEquals(2, allocatedContainers.size()); Assert.assertEquals(allocContainers + 2, metrics.getAllocatedContainers()); Assert.assertEquals(aggrAllocatedContainers + 2, metrics.getAggregatedAllocatedContainers()); Assert.assertEquals(aggrOffSwitchContainers + 2, metrics.getAggregatedOffSwitchContainers()); Container container = allocatedContainers.get(0); MockNM allocNode = nodes.get(container.getNodeId()); allocNode.nodeHeartbeat(Arrays.asList( ContainerStatus.newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.RUNNING, "", 0)), true); rm.drainEvents(); RMContainer rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( container.getId().getApplicationAttemptId()).getRMContainer( container.getId()); Assert.assertEquals(RMContainerState.RUNNING, rmContainer.getState()); allocNode.nodeHeartbeat(Arrays.asList( ContainerStatus.newInstance(container.getId(), ExecutionType.OPPORTUNISTIC, ContainerState.COMPLETE, "", 0)), true); rm.drainEvents(); rmContainer = ((CapacityScheduler) scheduler) .getApplicationAttempt( container.getId().getApplicationAttemptId()).getRMContainer( container.getId()); Assert.assertNull(rmContainer); Assert.assertEquals(allocContainers + 1, metrics.getAllocatedContainers()); Assert.assertEquals(aggrReleasedContainers + 1, metrics.getAggregatedReleasedContainers()); }