RMContainerImpl implements RMContainer { @Override public ApplicationAttemptId getApplicationAttemptId() { return this.appAttemptId; } 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 = 180000) public void testStoreAllContainerMetrics() throws Exception { Configuration conf = new Configuration(); conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 1); conf.setBoolean( YarnConfiguration.APPLICATION_HISTORY_SAVE_NON_AM_CONTAINER_META_INFO, true); MockRM rm1 = new MockRM(conf); SystemMetricsPublisher publisher = mock(SystemMetricsPublisher.class); rm1.getRMContext().setSystemMetricsPublisher(publisher); rm1.start(); MockNM nm1 = rm1.registerNode("unknownhost:1234", 8000); RMApp app1 = MockRMAppSubmitter.submitWithMemory(1024, rm1); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 1, ContainerState.RUNNING); am1.allocate("127.0.0.1", 1024, 1, new ArrayList<ContainerId>()); ContainerId containerId2 = ContainerId.newContainerId( am1.getApplicationAttemptId(), 2); rm1.waitForState(nm1, containerId2, RMContainerState.ALLOCATED); am1.allocate(new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>()) .getAllocatedContainers(); rm1.waitForState(nm1, containerId2, RMContainerState.ACQUIRED); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 2, ContainerState.RUNNING); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 2, ContainerState.COMPLETE); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 1, ContainerState.COMPLETE); rm1.waitForState(nm1, containerId2, RMContainerState.COMPLETED); rm1.stop(); verify(publisher, times(2)).containerCreated(any(RMContainer.class), anyLong()); verify(publisher, times(2)).containerFinished(any(RMContainer.class), anyLong()); }
@Test (timeout = 180000) public void testStoreOnlyAMContainerMetrics() throws Exception { Configuration conf = new Configuration(); conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 1); conf.setBoolean( YarnConfiguration.APPLICATION_HISTORY_SAVE_NON_AM_CONTAINER_META_INFO, false); MockRM rm1 = new MockRM(conf); SystemMetricsPublisher publisher = mock(SystemMetricsPublisher.class); rm1.getRMContext().setSystemMetricsPublisher(publisher); rm1.start(); MockNM nm1 = rm1.registerNode("unknownhost:1234", 8000); RMApp app1 = MockRMAppSubmitter.submitWithMemory(1024, rm1); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 1, ContainerState.RUNNING); am1.allocate("127.0.0.1", 1024, 1, new ArrayList<ContainerId>()); ContainerId containerId2 = ContainerId.newContainerId( am1.getApplicationAttemptId(), 2); rm1.waitForState(nm1, containerId2, RMContainerState.ALLOCATED); am1.allocate(new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>()) .getAllocatedContainers(); rm1.waitForState(nm1, containerId2, RMContainerState.ACQUIRED); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 2, ContainerState.RUNNING); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 2, ContainerState.COMPLETE); nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 1, ContainerState.COMPLETE); rm1.waitForState(nm1, containerId2, RMContainerState.COMPLETED); rm1.stop(); verify(publisher, times(1)).containerCreated(any(RMContainer.class), anyLong()); verify(publisher, times(1)).containerFinished(any(RMContainer.class), anyLong()); }