OpportunisticContainerAllocatorAMService extends ApplicationMasterService implements DistributedSchedulingAMProtocol,
    EventHandler<SchedulerEvent> { @Override public void handle(SchedulerEvent event) { switch (event.getType()) { case NODE_ADDED: if (!(event instanceof NodeAddedSchedulerEvent)) { throw new RuntimeException("Unexpected event type: " + event); } NodeAddedSchedulerEvent nodeAddedEvent = (NodeAddedSchedulerEvent) event; nodeMonitor.addNode(nodeAddedEvent.getContainerReports(), nodeAddedEvent.getAddedRMNode()); break; case NODE_REMOVED: if (!(event instanceof NodeRemovedSchedulerEvent)) { throw new RuntimeException("Unexpected event type: " + event); } NodeRemovedSchedulerEvent nodeRemovedEvent = (NodeRemovedSchedulerEvent) event; nodeMonitor.removeNode(nodeRemovedEvent.getRemovedRMNode()); break; case NODE_UPDATE: if (!(event instanceof NodeUpdateSchedulerEvent)) { throw new RuntimeException("Unexpected event type: " + event); } NodeUpdateSchedulerEvent nodeUpdatedEvent = (NodeUpdateSchedulerEvent) event; nodeMonitor.updateNode(nodeUpdatedEvent.getRMNode()); break; case NODE_RESOURCE_UPDATE: if (!(event instanceof NodeResourceUpdateSchedulerEvent)) { throw new RuntimeException("Unexpected event type: " + event); } NodeResourceUpdateSchedulerEvent nodeResourceUpdatedEvent = (NodeResourceUpdateSchedulerEvent) event; nodeMonitor.updateNodeResource(nodeResourceUpdatedEvent.getRMNode(), nodeResourceUpdatedEvent.getResourceOption()); break; case APP_ADDED: break; case APP_REMOVED: break; case APP_ATTEMPT_ADDED: break; case APP_ATTEMPT_REMOVED: break; case CONTAINER_EXPIRED: break; case NODE_LABELS_UPDATE: break; case RELEASE_CONTAINER: break; case NODE_ATTRIBUTES_UPDATE: break; case KILL_RESERVED_CONTAINER: break; case MARK_CONTAINER_FOR_PREEMPTION: break; case MARK_CONTAINER_FOR_KILLABLE: break; case MARK_CONTAINER_FOR_NONKILLABLE: break; case MANAGE_QUEUE: break; default: LOG.error("Unknown event arrived at" + "OpportunisticContainerAllocatorAMService: {}", event); } } 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 = 60000) public void testNodeRemovalDuringAllocate() throws Exception { MockNM nm1 = new MockNM("h1:1234", 4096, rm.getResourceTrackerService()); MockNM nm2 = new MockNM("h2:1234", 4096, rm.getResourceTrackerService()); 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); ApplicationAttemptId attemptId = app1.getCurrentAppAttempt().getAppAttemptId(); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm2); ResourceScheduler scheduler = rm.getResourceScheduler(); RMNode rmNode1 = rm.getRMContext().getRMNodes().get(nm1.getNodeId()); RMNode rmNode2 = rm.getRMContext().getRMNodes().get(nm2.getNodeId()); OpportunisticContainerContext ctxt = ((CapacityScheduler) scheduler) .getApplicationAttempt(attemptId).getOpportunisticContainerContext(); nm1.nodeHeartbeat(oppContainersStatus, true); nm2.nodeHeartbeat(oppContainersStatus, true); for (int i = 0; i < 10; i++) { am1.allocate( Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2)), null); if (ctxt.getNodeMap().size() == 2) { break; } Thread.sleep(50); } Assert.assertEquals(2, ctxt.getNodeMap().size()); scheduler.handle(new NodeRemovedSchedulerEvent(rmNode1)); for (int i = 0; i < 10; i++) { try { am1.allocate( Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(1 * GB), 2)), null); } catch (Exception e) { Assert.fail("Allocate request should be handled on node removal"); } if (ctxt.getNodeMap().size() == 1) { break; } Thread.sleep(50); } Assert.assertEquals(1, ctxt.getNodeMap().size()); }
@Test(timeout = 60000) public void testAppAttemptRemovalAfterNodeRemoval() throws Exception { MockNM nm = new MockNM("h:1234", 4096, rm.getResourceTrackerService()); nm.registerNode(); nm.nodeHeartbeat(oppContainersStatus, true); RMApp app = MockRMAppSubmitter.submit(rm, MockRMAppSubmissionData.Builder.createWithMemory(1 * GB, rm) .withAppName("app") .withUser("user") .withAcls(null) .withQueue("default") .build()); ApplicationAttemptId attemptId = app.getCurrentAppAttempt().getAppAttemptId(); MockAM am = MockRM.launchAndRegisterAM(app, rm, nm); ResourceScheduler scheduler = rm.getResourceScheduler(); SchedulerApplicationAttempt schedulerAttempt = ((CapacityScheduler)scheduler).getApplicationAttempt(attemptId); RMNode rmNode1 = rm.getRMContext().getRMNodes().get(nm.getNodeId()); nm.nodeHeartbeat(oppContainersStatus, true); GenericTestUtils.waitFor(() -> scheduler.getNumClusterNodes() == 1, 10, 200 * 100); AllocateResponse allocateResponse = am.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(); Container container = allocatedContainers.get(0); scheduler.handle(new NodeRemovedSchedulerEvent(rmNode1)); GenericTestUtils.waitFor(() -> scheduler.getNumClusterNodes() == 0, 10, 200 * 100); RMContainer rmContainer = null; rmContainer = SchedulerUtils.createOpportunisticRmContainer( rm.getRMContext(), container, true); if (rmContainer == null) { rmContainer = new RMContainerImpl(container, SchedulerRequestKey.extractFrom(container), schedulerAttempt.getApplicationAttemptId(), container.getNodeId(), schedulerAttempt.getUser(), rm.getRMContext(), true); } schedulerAttempt.addRMContainer(container.getId(), rmContainer); scheduler.handle(new AppAttemptRemovedSchedulerEvent(attemptId, RMAppAttemptState.FAILED, false)); }