@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;

    // <-- IGNORED EVENTS : START -->
    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;
    // <-- IGNORED EVENTS : END -->
    default:
      LOG.error("Unknown event arrived at" +
          "OpportunisticContainerAllocatorAMService: {}", event);
    }

  }
@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);

    //test YARN-9165
    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);
    }
    //test YARN-9164
    schedulerAttempt.addRMContainer(container.getId(), rmContainer);
    scheduler.handle(new AppAttemptRemovedSchedulerEvent(attemptId,
        RMAppAttemptState.FAILED, false));
  }