@Override
  public void handle(RMContainerEvent event) {
    LOG.debug("Processing {} of type {}", event.getContainerId(),
        event.getType());

    writeLock.lock();
    try {
      RMContainerState oldState = getState();
      try {
         stateMachine.doTransition(event.getType(), event);
      } catch (InvalidStateTransitionException e) {
        LOG.error("Can't handle this event at current state", e);
        onInvalidStateTransition(event.getType(), oldState);
      }
      if (oldState != getState()) {
        LOG.info(event.getContainerId() + " Container Transitioned from "
            + oldState + " to " + getState());
      }
    }
    
    finally {
      writeLock.unlock();
    }
  }
@Test(timeout = 30000)
  public void testContainerAcquiredAtKilled() {
    DrainDispatcher drainDispatcher = new DrainDispatcher();
    EventHandler<RMAppAttemptEvent> appAttemptEventHandler = mock(
        EventHandler.class);
    EventHandler generic = mock(EventHandler.class);
    drainDispatcher.register(RMAppAttemptEventType.class,
        appAttemptEventHandler);
    drainDispatcher.register(RMNodeEventType.class, generic);
    drainDispatcher.init(new YarnConfiguration());
    drainDispatcher.start();
    NodeId nodeId = BuilderUtils.newNodeId("host", 3425);
    ApplicationId appId = BuilderUtils.newApplicationId(1, 1);
    ApplicationAttemptId appAttemptId = BuilderUtils.newApplicationAttemptId(
        appId, 1);
    ContainerId containerId = BuilderUtils.newContainerId(appAttemptId, 1);
    ContainerAllocationExpirer expirer = mock(ContainerAllocationExpirer.class);

    Resource resource = Resources.createResource(512);
    Priority priority = BuilderUtils.newPriority(5);

    Container container = BuilderUtils.newContainer(containerId, nodeId,
        "host:3465", resource, priority, null);

    ConcurrentMap<ApplicationId, RMApp> appMap = new ConcurrentHashMap<>();
    RMApp rmApp = mock(RMApp.class);
    appMap.putIfAbsent(appId, rmApp);

    RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
    SystemMetricsPublisher publisher = mock(SystemMetricsPublisher.class);
    RMContext rmContext = mock(RMContext.class);
    when(rmContext.getDispatcher()).thenReturn(drainDispatcher);
    when(rmContext.getContainerAllocationExpirer()).thenReturn(expirer);
    when(rmContext.getRMApplicationHistoryWriter()).thenReturn(writer);
    when(rmContext.getSystemMetricsPublisher()).thenReturn(publisher);
    AllocationTagsManager ptm = mock(AllocationTagsManager.class);
    when(rmContext.getAllocationTagsManager()).thenReturn(ptm);

    YarnConfiguration conf = new YarnConfiguration();
    conf.setBoolean(
        YarnConfiguration.APPLICATION_HISTORY_SAVE_NON_AM_CONTAINER_META_INFO,
        true);
    when(rmContext.getYarnConfiguration()).thenReturn(conf);
    when(rmContext.getRMApps()).thenReturn(appMap);

    RMContainer rmContainer = new RMContainerImpl(container,
        SchedulerRequestKey.extractFrom(container), appAttemptId,
        nodeId, "user", rmContext) {
        @Override
        protected void onInvalidStateTransition(
            RMContainerEventType rmContainerEventType, RMContainerState state) {
            Assert.fail("RMContainerImpl: can't handle " + rmContainerEventType
                + " at state " + state);
        }
    };
    rmContainer.handle(new RMContainerEvent(containerId,
        RMContainerEventType.KILL));

    rmContainer.handle(new RMContainerEvent(containerId,
        RMContainerEventType.ACQUIRED));
  }