RMContainerImpl implements RMContainer { @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(); } } RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
      ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
      RMContext rmContext);  RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
      ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
      RMContext rmContext, boolean isExternallyAllocated);  RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
      ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
      RMContext rmContext, String nodeLabelExpression);  RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
      ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
      RMContext rmContext, long creationTime, String nodeLabelExpression);  RMContainerImpl(Container container, SchedulerRequestKey schedulerKey,
      ApplicationAttemptId appAttemptId, NodeId nodeId, String user,
      RMContext rmContext, long creationTime, String nodeLabelExpression,
      boolean isExternallyAllocated); @Override ContainerId getContainerId(); @Override ApplicationAttemptId getApplicationAttemptId(); @Override Container getContainer(); void setContainer(Container container); @Override RMContainerState getState(); @Override Resource getReservedResource(); @Override NodeId getReservedNode(); @Override SchedulerRequestKey getReservedSchedulerKey(); @Override Resource getAllocatedResource(); @Override Resource getLastConfirmedResource(); @Override NodeId getAllocatedNode(); @Override SchedulerRequestKey getAllocatedSchedulerKey(); @Override Priority getAllocatedPriority(); @Override long getCreationTime(); @Override long getFinishTime(); @Override String getDiagnosticsInfo(); @Override String getLogURL(); @Override int getContainerExitStatus(); @Override ContainerState getContainerState(); @Override ContainerRequest getContainerRequest(); void setContainerRequest(ContainerRequest request); @Override String toString(); @Override boolean isAMContainer(); void setAMContainer(boolean isAMContainer); @Override void handle(RMContainerEvent event); boolean completed(); @Override NodeId getNodeId(); @Override Set<String> getAllocationTags(); void setAllocationTags(Set<String> tags); @Override ContainerReport createContainerReport(); @Override String getNodeHttpAddress(); @Override Map<String, List<Map<String, String>>> getExposedPorts(); @Override void setExposedPorts(Map<String, List<Map<String, String>>> ports); @Override String getNodeLabelExpression(); @Override boolean equals(Object obj); @Override int hashCode(); @Override int compareTo(RMContainer o); void setQueueName(String queueName); @Override String getQueueName(); @Override ExecutionType getExecutionType(); @Override boolean isRemotelyAllocated(); @Override Resource getAllocatedOrReservedResource(); @Override void setContainerId(ContainerId containerId);  }
@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 = BuilderUtils.newResource(512, 1); 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)); }