Java 类org.apache.hadoop.yarn.api.records.ContainerState 实例源码

@Override
public StartContainersResponse startContainers(
    StartContainersRequest requests) throws YarnException {
  StartContainersResponse response =
      recordFactory.newRecordInstance(StartContainersResponse.class);
  for (StartContainerRequest request : requests.getStartContainerRequests()) {
    Token containerToken = request.getContainerToken();
    ContainerTokenIdentifier tokenId = null;

    try {
      tokenId = newContainerTokenIdentifier(containerToken);
    } catch (IOException e) {
      throw RPCUtil.getRemoteException(e);
    }
    ContainerStatus status =
        recordFactory.newRecordInstance(ContainerStatus.class);
    status.setState(ContainerState.RUNNING);
    status.setContainerId(tokenId.getContainerID());
    status.setExitStatus(0);
    statuses.add(status);

  }
  return response;
}

private void testGetContainerStatus(Container container, int index,
    ContainerState state, String diagnostics, List<Integer> exitStatuses)
        throws YarnException, IOException {
  while (true) {
    try {
      ContainerStatus status = nmClient.getContainerStatus(
          container.getId(), container.getNodeId());
      // NodeManager may still need some time to get the stable
      // container status
      if (status.getState() == state) {
        assertEquals(container.getId(), status.getContainerId());
        assertTrue("" + index + ": " + status.getDiagnostics(),
            status.getDiagnostics().contains(diagnostics));

        assertTrue("Exit Statuses are supposed to be in: " + exitStatuses +
            ", but the actual exit status code is: " + status.getExitStatus(),
            exitStatuses.contains(status.getExitStatus()));
        break;
      }
      Thread.sleep(100);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

private void waitForRecoveredContainers() throws InterruptedException {
  final int sleepMsec = 100;
  int waitIterations = 100;
  List<ContainerId> newContainers = new ArrayList<ContainerId>();
  while (--waitIterations >= 0) {
    newContainers.clear();
    for (Container container : context.getContainers().values()) {
      if (container.getContainerState() == org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState.NEW) {
        newContainers.add(container.getContainerId());
      }
    }
    if (newContainers.isEmpty()) {
      break;
    }
    LOG.info("Waiting for containers: " + newContainers);
    Thread.sleep(sleepMsec);
  }
  if (waitIterations < 0) {
    LOG.warn("Timeout waiting for recovered containers");
  }
}

private List<NMContainerStatus> getNMContainerStatuses() throws IOException {
  List<NMContainerStatus> containerStatuses =
      new ArrayList<NMContainerStatus>();
  for (Container container : this.context.getContainers().values()) {
    ContainerId containerId = container.getContainerId();
    ApplicationId applicationId = containerId.getApplicationAttemptId()
        .getApplicationId();
    if (!this.context.getApplications().containsKey(applicationId)) {
      context.getContainers().remove(containerId);
      continue;
    }
    NMContainerStatus status =
        container.getNMContainerStatus();
    containerStatuses.add(status);
    if (status.getContainerState() == ContainerState.COMPLETE) {
      // Adding to finished containers cache. Cache will keep it around at
      // least for #durationToTrackStoppedContainers duration. In the
      // subsequent call to stop container it will get removed from cache.
      addCompletedContainer(containerId);
    }
  }
  LOG.info("Sending out " + containerStatuses.size()
    + " NM container statuses: " + containerStatuses);
  return containerStatuses;
}

public static void waitForContainerState(ContainerManagementProtocol containerManager,
      ContainerId containerID, ContainerState finalState, int timeOutMax)
      throws InterruptedException, YarnException, IOException {
List<ContainerId> list = new ArrayList<ContainerId>();
list.add(containerID);
GetContainerStatusesRequest request =
    GetContainerStatusesRequest.newInstance(list);
ContainerStatus containerStatus =
    containerManager.getContainerStatuses(request).getContainerStatuses()
      .get(0);
int timeoutSecs = 0;
  while (!containerStatus.getState().equals(finalState)
      && timeoutSecs++ < timeOutMax) {
      Thread.sleep(1000);
      LOG.info("Waiting for container to get into state " + finalState
          + ". Current state is " + containerStatus.getState());
      containerStatus = containerManager.getContainerStatuses(request).getContainerStatuses().get(0);
    }
    LOG.info("Container state is " + containerStatus.getState());
    Assert.assertEquals("ContainerState is not correct (timedout)",
        finalState, containerStatus.getState());
  }

@Override
public ConcurrentMap<ContainerId, Container> getContainers() {
  if (heartBeatID == 0) {
    return containers;
  } else if (heartBeatID == 1) {
    ContainerStatus containerStatus2 =
        createContainerStatus(2, ContainerState.RUNNING);
    putMockContainer(containerStatus2);

    ContainerStatus containerStatus3 =
        createContainerStatus(3, ContainerState.COMPLETE);
    putMockContainer(containerStatus3);
    return containers;
  } else if (heartBeatID == 2) {
    ContainerStatus containerStatus4 =
        createContainerStatus(4, ContainerState.RUNNING);
    putMockContainer(containerStatus4);

    ContainerStatus containerStatus5 =
        createContainerStatus(5, ContainerState.COMPLETE);
    putMockContainer(containerStatus5);
    return containers;
  } else if (heartBeatID == 3 || heartBeatID == 4) {
    return containers;
  } else {
    containers.clear();
    return containers;
  }
}

private static RMContainer createRMContainer(ContainerId containerId) {
  RMContainer container = mock(RMContainer.class);
  when(container.getContainerId()).thenReturn(containerId);
  when(container.getAllocatedNode()).thenReturn(
      NodeId.newInstance("test host", -100));
  when(container.getAllocatedResource()).thenReturn(
      Resource.newInstance(-1, -1, -1));
  when(container.getAllocatedPriority()).thenReturn(Priority.UNDEFINED);
  when(container.getCreationTime()).thenReturn(Integer.MAX_VALUE + 1L);
  when(container.getFinishTime()).thenReturn(Integer.MAX_VALUE + 2L);
  when(container.getDiagnosticsInfo()).thenReturn("test diagnostics info");
  when(container.getContainerExitStatus()).thenReturn(-1);
  when(container.getContainerState()).thenReturn(ContainerState.COMPLETE);
  Container mockContainer = mock(Container.class);
  when(container.getContainer()).thenReturn(mockContainer);
  when(mockContainer.getNodeHttpAddress())
    .thenReturn("http://localhost:1234");
  return container;
}

@Test (timeout = 30000)
public void testAppWithNoContainers() throws Exception {
  Logger rootLogger = LogManager.getRootLogger();
  rootLogger.setLevel(Level.DEBUG);
  MockRM rm = new MockRM(conf);
  rm.start();
  MockNM nm1 = rm.registerNode("h1:1234", 5120);

  RMApp app = rm.submitApp(2000);

  //kick the scheduling
  nm1.nodeHeartbeat(true);

  RMAppAttempt attempt = app.getCurrentAppAttempt();
  MockAM am = rm.sendAMLaunched(attempt.getAppAttemptId());
  am.registerAppAttempt();
  am.unregisterAppAttempt();
  nm1.nodeHeartbeat(attempt.getAppAttemptId(), 1, ContainerState.COMPLETE);
  am.waitForState(RMAppAttemptState.FINISHED);
  rm.stop();
}

@Test (timeout = 60000)
public void testInvalidatedAMHostPortOnAMRestart() throws Exception {
  MockRM rm1 = new MockRM(conf);
  rm1.start();
  MockNM nm1 =
      new MockNM("127.0.0.1:1234", 15120, rm1.getResourceTrackerService());
  nm1.registerNode();

  // a failed app
  RMApp app2 = rm1.submitApp(200);
  MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);
  nm1
    .nodeHeartbeat(am2.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
  am2.waitForState(RMAppAttemptState.FAILED);
  rm1.waitForState(app2.getApplicationId(), RMAppState.ACCEPTED);

  // before new attempt is launched, the app report returns the invalid AM
  // host and port.
  GetApplicationReportRequest request1 =
      GetApplicationReportRequest.newInstance(app2.getApplicationId());
  ApplicationReport report1 =
      rm1.getClientRMService().getApplicationReport(request1)
        .getApplicationReport();
  Assert.assertEquals("N/A", report1.getHost());
  Assert.assertEquals(-1, report1.getRpcPort());
}

private static RMContainer createRMContainer(ContainerId containerId) {
  RMContainer container = mock(RMContainer.class);
  when(container.getContainerId()).thenReturn(containerId);
  when(container.getAllocatedNode()).thenReturn(
    NodeId.newInstance("test host", -100));
  when(container.getAllocatedResource()).thenReturn(
    Resource.newInstance(-1, -1, -1));
  when(container.getAllocatedPriority()).thenReturn(Priority.UNDEFINED);
  when(container.getCreationTime()).thenReturn(0L);
  when(container.getFinishTime()).thenReturn(1L);
  when(container.getDiagnosticsInfo()).thenReturn("test diagnostics info");
  when(container.getLogURL()).thenReturn("test log url");
  when(container.getContainerExitStatus()).thenReturn(-1);
  when(container.getContainerState()).thenReturn(ContainerState.COMPLETE);
  return container;
}

@Test
public void testSuccessfulFinishingToFinished() {
  Container amContainer = allocateApplicationAttempt();
  launchApplicationAttempt(amContainer);
  runApplicationAttempt(amContainer, "host", 8042, "oldtrackingurl", false);
  FinalApplicationStatus finalStatus = FinalApplicationStatus.SUCCEEDED;
  String trackingUrl = "mytrackingurl";
  String diagnostics = "Successful";
  unregisterApplicationAttempt(amContainer, finalStatus, trackingUrl,
      diagnostics);
  NodeId anyNodeId = NodeId.newInstance("host", 1234);
  applicationAttempt.handle(
      new RMAppAttemptContainerFinishedEvent(
          applicationAttempt.getAppAttemptId(),
          BuilderUtils.newContainerStatus(amContainer.getId(),
              ContainerState.COMPLETE, "", 0), anyNodeId));
  testAppAttemptFinishedState(amContainer, finalStatus, trackingUrl,
      diagnostics, 0, false);
}

public static List<NMContainerStatus>
    createNMContainerStatusForApp(MockAM am) {
  List<NMContainerStatus> list =
      new ArrayList<NMContainerStatus>();
  NMContainerStatus amContainer =
      TestRMRestart.createNMContainerStatus(am.getApplicationAttemptId(), 1,
        ContainerState.RUNNING);
  NMContainerStatus runningContainer =
      TestRMRestart.createNMContainerStatus(am.getApplicationAttemptId(), 2,
        ContainerState.RUNNING);
  NMContainerStatus completedContainer =
      TestRMRestart.createNMContainerStatus(am.getApplicationAttemptId(), 3,
        ContainerState.COMPLETE);
  list.add(amContainer);
  list.add(runningContainer);
  list.add(completedContainer);
  return list;
}

@Override
public ContainerState getContainerState() {
  ContainerReportProtoOrBuilder p = viaProto ? proto : builder;
  if (!p.hasContainerState()) {
    return null;
  }
  return convertFromProtoFormat(p.getContainerState());
}

@Override
public void setContainerState(ContainerState containerState) {
  maybeInitBuilder();
  if (containerState == null) {
    builder.clearContainerState();
    return;
  }
  builder.setContainerState(convertToProtoFormat(containerState));
}

@Override
public synchronized ContainerState getState() {
  ContainerStatusProtoOrBuilder p = viaProto ? proto : builder;
  if (!p.hasState()) {
    return null;
  }
  return convertFromProtoFormat(p.getState());
}

@Override
public synchronized void setState(ContainerState state) {
  maybeInitBuilder();
  if (state == null) {
    builder.clearState();
    return;
  }
  builder.setState(convertToProtoFormat(state));
}

public static NMContainerStatus newInstance(ContainerId containerId,
    ContainerState containerState, Resource allocatedResource,
    String diagnostics, int containerExitStatus, Priority priority,
    long creationTime) {
  NMContainerStatus status =
      Records.newRecord(NMContainerStatus.class);
  status.setContainerId(containerId);
  status.setContainerState(containerState);
  status.setAllocatedResource(allocatedResource);
  status.setDiagnostics(diagnostics);
  status.setContainerExitStatus(containerExitStatus);
  status.setPriority(priority);
  status.setCreationTime(creationTime);
  return status;
}

@Override
public ContainerState getContainerState() {
  NMContainerStatusProtoOrBuilder p = viaProto ? proto : builder;
  if (!p.hasContainerState()) {
    return null;
  }
  return convertFromProtoFormat(p.getContainerState());
}

/**
 * handle method for AMRMClientAsync.CallbackHandler container allocation
 *
 * @param status
 *            list of status
 */
private synchronized void onContainersCompleted(List<ContainerStatus> status) {
    Collection<ContainerId> failed = new java.util.LinkedList<ContainerId>();
    for (ContainerStatus s : status) {
        assert (s.getState().equals(ContainerState.COMPLETE));
        int exstatus = s.getExitStatus();
        TaskRecord r = runningTasks.get(s.getContainerId());
        if (r == null)
            continue;
        if (exstatus == ContainerExitStatus.SUCCESS) {
            finishedTasks.add(r);
            runningTasks.remove(s.getContainerId());
        } else {
            try {
                if (exstatus == ContainerExitStatus.class.getField(
                        "KILLED_EXCEEDED_PMEM").getInt(null)) {
                    this.abortJob("[DMLC] Task "
                            + r.taskId
                            + " killed because of exceeding allocated physical memory");
                    return;
                }
                if (exstatus == ContainerExitStatus.class.getField(
                        "KILLED_EXCEEDED_VMEM").getInt(null)) {
                    this.abortJob("[DMLC] Task "
                            + r.taskId
                            + " killed because of exceeding allocated virtual memory");
                    return;
                }
            } catch (Exception e) {
                LOG.warn(e.getMessage());
            }
            LOG.info("[DMLC] Task " + r.taskId + " exited with status "
                     + exstatus + " Diagnostics:"+ s.getDiagnostics());
            failed.add(s.getContainerId());
        }
    }
    //this.handleFailure(failed);
}

public static ContainerStatus newContainerStatus(ContainerId containerId,
    ContainerState containerState, String diagnostics, int exitStatus) {
  ContainerStatus containerStatus = recordFactory
    .newRecordInstance(ContainerStatus.class);
  containerStatus.setState(containerState);
  containerStatus.setContainerId(containerId);
  containerStatus.setDiagnostics(diagnostics);
  containerStatus.setExitStatus(exitStatus);
  return containerStatus;
}

@Test
public void testRegisterNodeManagerRequest() {
  RegisterNodeManagerRequest request =
      RegisterNodeManagerRequest.newInstance(
        NodeId.newInstance("host", 1234), 1234, Resource.newInstance(0, 0),
        "version", Arrays.asList(NMContainerStatus.newInstance(
          ContainerId.newContainerId(
            ApplicationAttemptId.newInstance(
              ApplicationId.newInstance(1234L, 1), 1), 1),
          ContainerState.RUNNING, Resource.newInstance(1024, 1), "good", -1,
          Priority.newInstance(0), 1234)), Arrays.asList(
          ApplicationId.newInstance(1234L, 1),
          ApplicationId.newInstance(1234L, 2)));

  // serialze to proto, and get request from proto
  RegisterNodeManagerRequest request1 =
      new RegisterNodeManagerRequestPBImpl(
          ((RegisterNodeManagerRequestPBImpl) request).getProto());

  // check values
  Assert.assertEquals(request1.getNMContainerStatuses().size(), request
      .getNMContainerStatuses().size());
  Assert.assertEquals(request1.getNMContainerStatuses().get(0).getContainerId(),
      request.getNMContainerStatuses().get(0).getContainerId());
  Assert.assertEquals(request1.getRunningApplications().size(), request
      .getRunningApplications().size());
  Assert.assertEquals(request1.getRunningApplications().get(0), request
      .getRunningApplications().get(0));
  Assert.assertEquals(request1.getRunningApplications().get(1), request
      .getRunningApplications().get(1));
}

@Test
public void testRegisterNodeManagerRequest() {
  ApplicationId appId = ApplicationId.newInstance(123456789, 1);
  ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(appId, 1);
  ContainerId containerId = ContainerId.newContainerId(attemptId, 1);

  NMContainerStatus containerReport =
      NMContainerStatus.newInstance(containerId,
        ContainerState.RUNNING, Resource.newInstance(1024, 1, 2), "diagnostics",
        0, Priority.newInstance(10), 1234);
  List<NMContainerStatus> reports = Arrays.asList(containerReport);
  RegisterNodeManagerRequest request =
      RegisterNodeManagerRequest.newInstance(
        NodeId.newInstance("1.1.1.1", 1000), 8080,
          Resource.newInstance(1024, 1, 2), "NM-version-id", reports,
          Arrays.asList(appId));
  RegisterNodeManagerRequest requestProto =
      new RegisterNodeManagerRequestPBImpl(
        ((RegisterNodeManagerRequestPBImpl) request).getProto());
  Assert.assertEquals(containerReport, requestProto
    .getNMContainerStatuses().get(0));
  Assert.assertEquals(8080, requestProto.getHttpPort());
  Assert.assertEquals("NM-version-id", requestProto.getNMVersion());
  Assert.assertEquals(NodeId.newInstance("1.1.1.1", 1000),
    requestProto.getNodeId());
  Assert.assertEquals(Resource.newInstance(1024, 1, 2),
    requestProto.getResource());
  Assert.assertEquals(1, requestProto.getRunningApplications().size());
  Assert.assertEquals(appId, requestProto.getRunningApplications().get(0)); 
}

@VisibleForTesting
protected List<ContainerStatus> getContainerStatuses() throws IOException {
  List<ContainerStatus> containerStatuses = new ArrayList<ContainerStatus>();
  for (Container container : this.context.getContainers().values()) {
    ContainerId containerId = container.getContainerId();
    ApplicationId applicationId = containerId.getApplicationAttemptId()
        .getApplicationId();
    org.apache.hadoop.yarn.api.records.ContainerStatus containerStatus =
        container.cloneAndGetContainerStatus();
    if (containerStatus.getState() == ContainerState.COMPLETE) {
      if (isApplicationStopped(applicationId)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(applicationId + " is completing, " + " remove "
              + containerId + " from NM context.");
        }
        context.getContainers().remove(containerId);
        pendingCompletedContainers.put(containerId, containerStatus);
      } else {
        if (!isContainerRecentlyStopped(containerId)) {
          pendingCompletedContainers.put(containerId, containerStatus);
          // Adding to finished containers cache. Cache will keep it around at
          // least for #durationToTrackStoppedContainers duration. In the
          // subsequent call to stop container it will get removed from cache.
          addCompletedContainer(containerId);
        }
      }
    } else {
      containerStatuses.add(containerStatus);
    }
  }
  containerStatuses.addAll(pendingCompletedContainers.values());
  if (LOG.isDebugEnabled()) {
    LOG.debug("Sending out " + containerStatuses.size()
        + " container statuses: " + containerStatuses);
  }
  return containerStatuses;
}

public static ContainerStatus createContainerStatus(int id,
    ContainerState containerState) {
  ApplicationId applicationId = ApplicationId.newInstance(0, 1);
  ApplicationAttemptId applicationAttemptId =
      ApplicationAttemptId.newInstance(applicationId, 1);
  ContainerId contaierId = ContainerId.newContainerId(applicationAttemptId, id);
  ContainerStatus containerStatus =
      BuilderUtils.newContainerStatus(contaierId, containerState,
        "test_containerStatus: id=" + id + ", containerState: "
            + containerState, 0);
  return containerStatus;
}

public static Container getMockContainer(ContainerStatus containerStatus) {
  ContainerImpl container = mock(ContainerImpl.class);
  when(container.cloneAndGetContainerStatus()).thenReturn(containerStatus);
  when(container.getCurrentState()).thenReturn(containerStatus.getState());
  when(container.getContainerId()).thenReturn(
    containerStatus.getContainerId());
  if (containerStatus.getState().equals(ContainerState.COMPLETE)) {
    when(container.getContainerState())
      .thenReturn(org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState.DONE);
  } else if (containerStatus.getState().equals(ContainerState.RUNNING)) {
    when(container.getContainerState())
    .thenReturn(org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState.RUNNING);
  }
  return container;
}

@Test(timeout = 10000)
public void testGetContainers() throws YarnException, IOException {
  Configuration conf = new Configuration();
  conf.setBoolean(YarnConfiguration.APPLICATION_HISTORY_ENABLED,
      true);

  final YarnClient client = new MockYarnClient();
  client.init(conf);
  client.start();

  ApplicationId applicationId = ApplicationId.newInstance(1234, 5);
  ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(
      applicationId, 1);
  List<ContainerReport> reports = client.getContainers(appAttemptId);
  Assert.assertNotNull(reports);
  Assert.assertEquals(reports.get(0).getContainerId(),
      (ContainerId.newContainerId(appAttemptId, 1)));
  Assert.assertEquals(reports.get(1).getContainerId(),
      (ContainerId.newContainerId(appAttemptId, 2)));
  Assert.assertEquals(reports.get(2).getContainerId(),
      (ContainerId.newContainerId(appAttemptId, 3)));

  //First2 containers should come from RM with updated state information and 
  // 3rd container is not there in RM and should
  Assert.assertEquals(ContainerState.RUNNING,
      (reports.get(0).getContainerState()));
  Assert.assertEquals(ContainerState.RUNNING,
      (reports.get(1).getContainerState()));
  Assert.assertEquals(ContainerState.COMPLETE,
      (reports.get(2).getContainerState()));
  client.stop();
}

/**
 * Helper method to handle received ContainerStatus. If this corresponds to
 * the completion of a master-container of a managed AM,
 * we call the handler for RMAppAttemptContainerFinishedEvent.
 */
@SuppressWarnings("unchecked")
@VisibleForTesting
void handleNMContainerStatus(NMContainerStatus containerStatus, NodeId nodeId) {
  ApplicationAttemptId appAttemptId =
      containerStatus.getContainerId().getApplicationAttemptId();
  RMApp rmApp =
      rmContext.getRMApps().get(appAttemptId.getApplicationId());
  if (rmApp == null) {
    LOG.error("Received finished container : "
        + containerStatus.getContainerId()
        + " for unknown application " + appAttemptId.getApplicationId()
        + " Skipping.");
    return;
  }

  if (rmApp.getApplicationSubmissionContext().getUnmanagedAM()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Ignoring container completion status for unmanaged AM "
          + rmApp.getApplicationId());
    }
    return;
  }

  RMAppAttempt rmAppAttempt = rmApp.getRMAppAttempt(appAttemptId);
  Container masterContainer = rmAppAttempt.getMasterContainer();
  if (masterContainer.getId().equals(containerStatus.getContainerId())
      && containerStatus.getContainerState() == ContainerState.COMPLETE) {
    ContainerStatus status =
        ContainerStatus.newInstance(containerStatus.getContainerId(),
          containerStatus.getContainerState(), containerStatus.getDiagnostics(),
          containerStatus.getContainerExitStatus());
    // sending master container finished event.
    RMAppAttemptContainerFinishedEvent evt =
        new RMAppAttemptContainerFinishedEvent(appAttemptId, status,
            nodeId);
    rmContext.getDispatcher().getEventHandler().handle(evt);
  }
}

@Override
public ContainerState getContainerState() {
  try {
    readLock.lock();
    if (getFinishedStatus() != null) {
      return getFinishedStatus().getState();
    } else {
      return ContainerState.RUNNING;
    }
  } finally {
    readLock.unlock();
  }
}

@Override
public RMContainerState transition(RMContainerImpl container,
    RMContainerEvent event) {
  NMContainerStatus report =
      ((RMContainerRecoverEvent) event).getContainerReport();
  if (report.getContainerState().equals(ContainerState.COMPLETE)) {
    ContainerStatus status =
        ContainerStatus.newInstance(report.getContainerId(),
          report.getContainerState(), report.getDiagnostics(),
          report.getContainerExitStatus());

    new FinishedTransition().transition(container,
      new RMContainerFinishedEvent(container.containerId, status,
        RMContainerEventType.FINISHED));
    return RMContainerState.COMPLETED;
  } else if (report.getContainerState().equals(ContainerState.RUNNING)) {
    // Tell the app
    container.eventHandler.handle(new RMAppRunningOnNodeEvent(container
        .getApplicationAttemptId().getApplicationId(), container.nodeId));
    return RMContainerState.RUNNING;
  } else {
    // This can never happen.
    LOG.warn("RMContainer received unexpected recover event with container"
        + " state " + report.getContainerState() + " while recovering.");
    return RMContainerState.RUNNING;
  }
}

public ContainerFinishedEvent(
    ContainerId containerId,
    String diagnosticsInfo,
    int containerExitStatus,
    ContainerState state,
    long finishedTime) {
  super(SystemMetricsEventType.CONTAINER_FINISHED, finishedTime);
  this.containerId = containerId;
  this.diagnosticsInfo = diagnosticsInfo;
  this.containerExitStatus = containerExitStatus;
  this.state = state;
}

/**
 * Utility to create a {@link ContainerStatus} during exceptional
 * circumstances.
 * 
 * @param containerId {@link ContainerId} of returned/released/lost container.
 * @param diagnostics diagnostic message
 * @return <code>ContainerStatus</code> for an returned/released/lost 
 *         container
 */
private static ContainerStatus createAbnormalContainerStatus(
    ContainerId containerId, int exitStatus, String diagnostics) {
  ContainerStatus containerStatus = 
      recordFactory.newRecordInstance(ContainerStatus.class);
  containerStatus.setContainerId(containerId);
  containerStatus.setDiagnostics(diagnostics);
  containerStatus.setExitStatus(exitStatus);
  containerStatus.setState(ContainerState.COMPLETE);
  return containerStatus;
}

private void killOrphanContainerOnNode(RMNode node,
    NMContainerStatus container) {
  if (!container.getContainerState().equals(ContainerState.COMPLETE)) {
    this.rmContext.getDispatcher().getEventHandler().handle(
      new RMNodeCleanContainerEvent(node.getNodeID(),
        container.getContainerId()));
  }
}

/**
 * Process node labels update on a node.
 * 
 * TODO: Currently capacity scheduler will kill containers on a node when
 * labels on the node changed. It is a simply solution to ensure guaranteed
 * capacity on labels of queues. When YARN-2498 completed, we can let
 * preemption policy to decide if such containers need to be killed or just
 * keep them running.
 */
private synchronized void updateLabelsOnNode(NodeId nodeId,
    Set<String> newLabels) {
  FiCaSchedulerNode node = nodes.get(nodeId);
  if (null == node) {
    return;
  }

  // labels is same, we don't need do update
  if (node.getLabels().size() == newLabels.size()
      && node.getLabels().containsAll(newLabels)) {
    return;
  }

  // Kill running containers since label is changed
  for (RMContainer rmContainer : node.getRunningContainers()) {
    ContainerId containerId = rmContainer.getContainerId();
    completedContainer(rmContainer, 
        ContainerStatus.newInstance(containerId,
            ContainerState.COMPLETE, 
            String.format(
                "Container=%s killed since labels on the node=%s changed",
                containerId.toString(), nodeId.toString()),
            ContainerExitStatus.KILLED_BY_RESOURCEMANAGER),
        RMContainerEventType.KILL);
  }

  // Unreserve container on this node
  RMContainer reservedContainer = node.getReservedContainer();
  if (null != reservedContainer) {
    dropContainerReservation(reservedContainer);
  }

  // Update node labels after we've done this
  node.updateLabels(newLabels);
}

@Test
public void testAppsQueryFinishEnd() throws JSONException, Exception {
  rm.start();
  MockNM amNodeManager = rm.registerNode("127.0.0.1:1234", 2048);
  RMApp app1 = rm.submitApp(CONTAINER_MB);
  amNodeManager.nodeHeartbeat(true);
  // finish App
  MockAM am = rm
      .sendAMLaunched(app1.getCurrentAppAttempt().getAppAttemptId());
  am.registerAppAttempt();
  am.unregisterAppAttempt();
  amNodeManager.nodeHeartbeat(app1.getCurrentAppAttempt().getAppAttemptId(),
      1, ContainerState.COMPLETE);

  rm.submitApp(CONTAINER_MB);
  rm.submitApp(CONTAINER_MB);
  long end = System.currentTimeMillis();

  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("cluster")
      .path("apps").queryParam("finishedTimeEnd", String.valueOf(end))
      .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
  JSONObject json = response.getEntity(JSONObject.class);
  assertEquals("incorrect number of elements", 1, json.length());
  JSONObject apps = json.getJSONObject("apps");
  assertEquals("incorrect number of elements", 1, apps.length());
  JSONArray array = apps.getJSONArray("app");
  assertEquals("incorrect number of elements", 3, array.length());
  rm.stop();
}

private void finishTask(DrainDispatcher rmDispatcher, MockNM node,
    MRApp mrApp, Task task) throws Exception {
  TaskAttempt attempt = task.getAttempts().values().iterator().next();
  List<ContainerStatus> contStatus = new ArrayList<ContainerStatus>(1);
  contStatus.add(ContainerStatus.newInstance(attempt.getAssignedContainerID(),
      ContainerState.COMPLETE, "", 0));
  Map<ApplicationId,List<ContainerStatus>> statusUpdate =
      new HashMap<ApplicationId,List<ContainerStatus>>(1);
  statusUpdate.put(mrApp.getAppID(), contStatus);
  node.nodeHeartbeat(statusUpdate, true);
  rmDispatcher.await();
  mrApp.getContext().getEventHandler().handle(
        new TaskAttemptEvent(attempt.getID(), TaskAttemptEventType.TA_DONE));
  mrApp.waitForState(task, TaskState.SUCCEEDED);
}

@Public
@Unstable
public static ContainerFinishData newInstance(ContainerId containerId,
    long finishTime, String diagnosticsInfo, int containerExitCode,
    ContainerState containerState) {
  ContainerFinishData containerFD =
      Records.newRecord(ContainerFinishData.class);
  containerFD.setContainerId(containerId);
  containerFD.setFinishTime(finishTime);
  containerFD.setDiagnosticsInfo(diagnosticsInfo);
  containerFD.setContainerExitStatus(containerExitCode);
  containerFD.setContainerState(containerState);
  return containerFD;
}

@Test
public void testGetContainerReport() throws Exception {
  ApplicationCLI cli = createAndGetAppCLI();
  ApplicationId applicationId = ApplicationId.newInstance(1234, 5);
  ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
      applicationId, 1);
  ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
  ContainerReport container = ContainerReport.newInstance(containerId, null,
      NodeId.newInstance("host", 1234), Priority.UNDEFINED, 1234, 5678,
      "diagnosticInfo", "logURL", 0, ContainerState.COMPLETE,
      "http://" + NodeId.newInstance("host", 2345).toString());
  when(client.getContainerReport(any(ContainerId.class))).thenReturn(
      container);
  int result = cli.run(new String[] { "container", "-status",
      containerId.toString() });
  assertEquals(0, result);
  verify(client).getContainerReport(containerId);
  ByteArrayOutputStream baos = new ByteArrayOutputStream();
  PrintWriter pw = new PrintWriter(baos);
  pw.println("Container Report : ");
  pw.println("\tContainer-Id : container_1234_0005_01_000001");
  pw.println("\tStart-Time : 1234");
  pw.println("\tFinish-Time : 5678");
  pw.println("\tState : COMPLETE");
  pw.println("\tLOG-URL : logURL");
  pw.println("\tHost : host:1234");
  pw.println("\tNodeHttpAddress : http://host:2345");
  pw.println("\tDiagnostics : diagnosticInfo");
  pw.close();
  String appReportStr = baos.toString("UTF-8");
  Assert.assertEquals(appReportStr, sysOutStream.toString());
  verify(sysOut, times(1)).println(isA(String.class));
}

public List<UpdatedContainerInfo> pullContainerUpdates() {
  ArrayList<UpdatedContainerInfo> list = new ArrayList<UpdatedContainerInfo>();

  ArrayList<ContainerStatus> list2 = new ArrayList<ContainerStatus>();
  for(ContainerId cId : this.toCleanUpContainers) {
    list2.add(ContainerStatus.newInstance(cId, ContainerState.RUNNING, "", 
      ContainerExitStatus.SUCCESS));
  }
  list.add(new UpdatedContainerInfo(new ArrayList<ContainerStatus>(), 
    list2));
  return list;
}

@Test
public void testRunningToFailed() {
  Container amContainer = allocateApplicationAttempt();
  launchApplicationAttempt(amContainer);
  runApplicationAttempt(amContainer, "host", 8042, "oldtrackingurl", false);
  String containerDiagMsg = "some error";
  int exitCode = 123;
  ContainerStatus cs = BuilderUtils.newContainerStatus(amContainer.getId(),
      ContainerState.COMPLETE, containerDiagMsg, exitCode);
  ApplicationAttemptId appAttemptId = applicationAttempt.getAppAttemptId();
  NodeId anyNodeId = NodeId.newInstance("host", 1234);
  applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
      appAttemptId, cs, anyNodeId));

  // ignored ContainerFinished and Expire at FinalSaving if we were supposed
  // to Failed state.
  assertEquals(RMAppAttemptState.FINAL_SAVING,
    applicationAttempt.getAppAttemptState());
  applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
    applicationAttempt.getAppAttemptId(), BuilderUtils.newContainerStatus(
      amContainer.getId(), ContainerState.COMPLETE, "", 0), anyNodeId));
  applicationAttempt.handle(new RMAppAttemptEvent(
    applicationAttempt.getAppAttemptId(), RMAppAttemptEventType.EXPIRE));
  assertEquals(RMAppAttemptState.FINAL_SAVING,
    applicationAttempt.getAppAttemptState()); 
  assertEquals(YarnApplicationAttemptState.RUNNING,
      applicationAttempt.createApplicationAttemptState());
  sendAttemptUpdateSavedEvent(applicationAttempt);
  assertEquals(RMAppAttemptState.FAILED,
      applicationAttempt.getAppAttemptState());
  assertEquals(0, applicationAttempt.getJustFinishedContainers().size());
  assertEquals(amContainer, applicationAttempt.getMasterContainer());
  assertEquals(0, application.getRanNodes().size());
  String rmAppPageUrl = pjoin(RM_WEBAPP_ADDR, "cluster", "app",
      applicationAttempt.getAppAttemptId().getApplicationId());
  assertEquals(rmAppPageUrl, applicationAttempt.getOriginalTrackingUrl());
  assertEquals(rmAppPageUrl, applicationAttempt.getTrackingUrl());
  verifyAMHostAndPortInvalidated();
  verifyApplicationAttemptFinished(RMAppAttemptState.FAILED);
}

@Test
public void testRunningToKilled() {
  Container amContainer = allocateApplicationAttempt();
  launchApplicationAttempt(amContainer);
  runApplicationAttempt(amContainer, "host", 8042, "oldtrackingurl", false);
  applicationAttempt.handle(
      new RMAppAttemptEvent(
          applicationAttempt.getAppAttemptId(),
          RMAppAttemptEventType.KILL));

  // ignored ContainerFinished and Expire at FinalSaving if we were supposed
  // to Killed state.
  assertEquals(RMAppAttemptState.FINAL_SAVING,
    applicationAttempt.getAppAttemptState());
  NodeId anyNodeId = NodeId.newInstance("host", 1234);
  applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
    applicationAttempt.getAppAttemptId(), BuilderUtils.newContainerStatus(
      amContainer.getId(), ContainerState.COMPLETE, "", 0), anyNodeId));
  applicationAttempt.handle(new RMAppAttemptEvent(
    applicationAttempt.getAppAttemptId(), RMAppAttemptEventType.EXPIRE));
  assertEquals(RMAppAttemptState.FINAL_SAVING,
    applicationAttempt.getAppAttemptState()); 
  assertEquals(YarnApplicationAttemptState.RUNNING,
      applicationAttempt.createApplicationAttemptState());
  sendAttemptUpdateSavedEvent(applicationAttempt);
  assertEquals(RMAppAttemptState.KILLED,
      applicationAttempt.getAppAttemptState());
  assertEquals(0, applicationAttempt.getJustFinishedContainers().size());
  assertEquals(amContainer, applicationAttempt.getMasterContainer());
  assertEquals(0, application.getRanNodes().size());
  String rmAppPageUrl = pjoin(RM_WEBAPP_ADDR, "cluster", "app",
      applicationAttempt.getAppAttemptId().getApplicationId());
  assertEquals(rmAppPageUrl, applicationAttempt.getOriginalTrackingUrl());
  assertEquals(rmAppPageUrl, applicationAttempt.getTrackingUrl());
  verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
  verifyAMHostAndPortInvalidated();
  verifyApplicationAttemptFinished(RMAppAttemptState.KILLED);
}