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

@Test (timeout = 10000)
public void testAMRMClientAsyncShutDown() throws Exception {
  Configuration conf = new Configuration();
  TestCallbackHandler callbackHandler = new TestCallbackHandler();
  @SuppressWarnings("unchecked")
  AMRMClient<ContainerRequest> client = mock(AMRMClientImpl.class);

  createAllocateResponse(new ArrayList<ContainerStatus>(),
    new ArrayList<Container>(), null);
  when(client.allocate(anyFloat())).thenThrow(
    new ApplicationAttemptNotFoundException("app not found, shut down"));

  AMRMClientAsync<ContainerRequest> asyncClient =
      AMRMClientAsync.createAMRMClientAsync(client, 10, callbackHandler);
  asyncClient.init(conf);
  asyncClient.start();

  asyncClient.registerApplicationMaster("localhost", 1234, null);

  Thread.sleep(50);

  verify(client, times(1)).allocate(anyFloat());
  asyncClient.stop();
}

private static void publishContainerEndEvent(
    final TimelineClient timelineClient, ContainerStatus container,
    String domainId, UserGroupInformation ugi) {
  final TimelineEntity entity = new TimelineEntity();
  entity.setEntityId(container.getContainerId().toString());
  entity.setEntityType(DSEntity.DS_CONTAINER.toString());
  entity.setDomainId(domainId);
  entity.addPrimaryFilter("user", ugi.getShortUserName());
  TimelineEvent event = new TimelineEvent();
  event.setTimestamp(System.currentTimeMillis());
  event.setEventType(DSEvent.DS_CONTAINER_END.toString());
  event.addEventInfo("State", container.getState().name());
  event.addEventInfo("Exit Status", container.getExitStatus());
  entity.addEvent(event);
  try {
    timelineClient.putEntities(entity);
  } catch (YarnException | IOException e) {
    LOG.error("Container end event could not be published for "
        + container.getContainerId().toString(), e);
  }
}

@Test
public void testAMCrashAtAllocated() {
  Container amContainer = allocateApplicationAttempt();
  String containerDiagMsg = "some error";
  int exitCode = 123;
  ContainerStatus cs =
      BuilderUtils.newContainerStatus(amContainer.getId(),
        ContainerState.COMPLETE, containerDiagMsg, exitCode);
  NodeId anyNodeId = NodeId.newInstance("host", 1234);
  applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
    applicationAttempt.getAppAttemptId(), cs, anyNodeId));
  assertEquals(YarnApplicationAttemptState.ALLOCATED,
      applicationAttempt.createApplicationAttemptState());
  sendAttemptUpdateSavedEvent(applicationAttempt);
  assertEquals(RMAppAttemptState.FAILED,
    applicationAttempt.getAppAttemptState());
  verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
  verifyApplicationAttemptFinished(RMAppAttemptState.FAILED);
  boolean shouldCheckURL = (applicationAttempt.getTrackingUrl() != null);
  verifyAMCrashAtAllocatedDiagnosticInfo(applicationAttempt.getDiagnostics(),
    exitCode, shouldCheckURL);
}

@SuppressWarnings("deprecation")
@Override
public void onContainerStatusReceived(ContainerId containerId,
    ContainerStatus containerStatus) {
  if (containerId.getId() >= expectedSuccess) {
    errorMsgs.add("Container " + containerId +
        " should throw the exception onContainerStatusReceived");
    return;
  }
  actualQuerySuccess.addAndGet(1);
  actualQuerySuccessArray.set(containerId.getId(), 1);
  // move on to the following success tests
  asyncClient.stopContainerAsync(containerId, nodeId);

  // Shouldn't crash the test thread
  throw new RuntimeException("Ignorable Exception");
}

public static AllocateResponse newAllocateResponse(int responseId,
    List<ContainerStatus> completedContainers,
    List<Container> allocatedContainers, List<NodeReport> updatedNodes,
    Resource availResources, AMCommand command, int numClusterNodes,
    PreemptionMessage preempt) {
  AllocateResponse response = recordFactory
      .newRecordInstance(AllocateResponse.class);
  response.setNumClusterNodes(numClusterNodes);
  response.setResponseId(responseId);
  response.setCompletedContainersStatuses(completedContainers);
  response.setAllocatedContainers(allocatedContainers);
  response.setUpdatedNodes(updatedNodes);
  response.setAvailableResources(availResources);
  response.setAMCommand(command);
  response.setPreemptionMessage(preempt);

  return response;
}

private NodeStatus getNodeStatus() {
  NodeStatus status = recordFactory.newRecordInstance(NodeStatus.class);
  status.setContainersStatuses(new ArrayList<ContainerStatus>());
  status.setKeepAliveApplications(new ArrayList<ApplicationId>());

  status.setNodeHealthStatus(getNodeHealthStatus());
  status.setNodeId(getNodeId());
  status.setResponseId(1);
  return status;
}

/**
 * Get a list of container statuses running on this NodeManager
 */
@Override
public GetContainerStatusesResponse getContainerStatuses(
    GetContainerStatusesRequest request) throws YarnException, IOException {

  List<ContainerStatus> succeededRequests = new ArrayList<ContainerStatus>();
  Map<ContainerId, SerializedException> failedRequests =
      new HashMap<ContainerId, SerializedException>();
  UserGroupInformation remoteUgi = getRemoteUgi();
  NMTokenIdentifier identifier = selectNMTokenIdentifier(remoteUgi);
  for (ContainerId id : request.getContainerIds()) {
    try {
      ContainerStatus status = getContainerStatusInternal(id, identifier);
      succeededRequests.add(status);
    } catch (YarnException e) {
      failedRequests.put(id, SerializedException.newInstance(e));
    }
  }
  return GetContainerStatusesResponse.newInstance(succeededRequests,
    failedRequests);
}

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

@Override
synchronized public GetContainerStatusesResponse getContainerStatuses(
    GetContainerStatusesRequest request) throws YarnException {
  List<ContainerStatus> statuses = new ArrayList<ContainerStatus>();
  for (ContainerId containerId : request.getContainerIds()) {
    List<Container> appContainers =
        containers.get(containerId.getApplicationAttemptId()
          .getApplicationId());
    Container container = null;
    for (Container c : appContainers) {
      if (c.getId().equals(containerId)) {
        container = c;
      }
    }
    if (container != null
        && containerStatusMap.get(container).getState() != null) {
      statuses.add(containerStatusMap.get(container));
    }
  }
  return GetContainerStatusesResponse.newInstance(statuses, null);
}

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

private Map<ApplicationId, List<ContainerStatus>> getAppToContainerStatusMap(
    List<ContainerStatus> containers) {
  Map<ApplicationId, List<ContainerStatus>> map =
      new HashMap<ApplicationId, List<ContainerStatus>>();
  for (ContainerStatus cs : containers) {
    ApplicationId applicationId =
        cs.getContainerId().getApplicationAttemptId().getApplicationId();
    List<ContainerStatus> appContainers = map.get(applicationId);
    if (appContainers == null) {
      appContainers = new ArrayList<ContainerStatus>();
      map.put(applicationId, appContainers);
    }
    appContainers.add(cs);
  }
  return map;
}

@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
@Unstable
public static AllocateResponse newInstance(int responseId,
    List<ContainerStatus> completedContainers,
    List<Container> allocatedContainers, List<NodeReport> updatedNodes,
    Resource availResources, AMCommand command, int numClusterNodes,
    PreemptionMessage preempt, List<NMToken> nmTokens, Token amRMToken,
    List<ContainerResourceIncrease> increasedContainers,
    List<ContainerResourceDecrease> decreasedContainers) {
  AllocateResponse response =
      newInstance(responseId, completedContainers, allocatedContainers,
        updatedNodes, availResources, command, numClusterNodes, preempt,
        nmTokens, increasedContainers, decreasedContainers);
  response.setAMRMToken(amRMToken);
  return response;
}

private String getAMContainerCrashedDiagnostics(
    RMAppAttemptContainerFinishedEvent finishEvent) {
  ContainerStatus status = finishEvent.getContainerStatus();
  StringBuilder diagnosticsBuilder = new StringBuilder();
  diagnosticsBuilder.append("AM Container for ").append(
    finishEvent.getApplicationAttemptId()).append(
    " exited with ").append(" exitCode: ").append(status.getExitStatus()).
    append("\n");
  if (this.getTrackingUrl() != null) {
    diagnosticsBuilder.append("For more detailed output,").append(
      " check application tracking page:").append(
      this.getTrackingUrl()).append(
      "Then, click on links to logs of each attempt.\n");
  }
  diagnosticsBuilder.append("Diagnostics: ").append(status.getDiagnostics())
      .append("Failing this attempt");
  return diagnosticsBuilder.toString();
}

@Override
public RMAppAttemptState transition(RMAppAttemptImpl appAttempt,
    RMAppAttemptEvent event) {

  RMAppAttemptContainerFinishedEvent containerFinishedEvent =
      (RMAppAttemptContainerFinishedEvent) event;
  ContainerStatus containerStatus =
      containerFinishedEvent.getContainerStatus();

  // Is this container the AmContainer? If the finished container is same as
  // the AMContainer, AppAttempt fails
  if (appAttempt.masterContainer != null
      && appAttempt.masterContainer.getId().equals(
          containerStatus.getContainerId())) {
    appAttempt.sendAMContainerToNM(appAttempt, containerFinishedEvent);

    // Remember the follow up transition and save the final attempt state.
    appAttempt.rememberTargetTransitionsAndStoreState(event,
        transitionToDo, RMAppAttemptState.FAILED, RMAppAttemptState.FAILED);
    return RMAppAttemptState.FINAL_SAVING;
  }

  // Add all finished containers so that they can be acked to NM
  addJustFinishedContainer(appAttempt, containerFinishedEvent);
  return this.currentState;
}

@Override
public RMAppAttemptState transition(RMAppAttemptImpl appAttempt,
    RMAppAttemptEvent event) {

  RMAppAttemptContainerFinishedEvent containerFinishedEvent
    = (RMAppAttemptContainerFinishedEvent) event;
  ContainerStatus containerStatus =
      containerFinishedEvent.getContainerStatus();

  // Is this container the ApplicationMaster container?
  if (appAttempt.masterContainer.getId().equals(
      containerStatus.getContainerId())) {
    new FinalTransition(RMAppAttemptState.FINISHED).transition(
        appAttempt, containerFinishedEvent);
    appAttempt.sendAMContainerToNM(appAttempt, containerFinishedEvent);
    return RMAppAttemptState.FINISHED;
  }
  // Add all finished containers so that they can be acked to NM.
  addJustFinishedContainer(appAttempt, containerFinishedEvent);

  return RMAppAttemptState.FINISHING;
}

@Override
public ContainerStatus getContainerStatus(ContainerId containerId,
    NodeId nodeId) throws YarnException, IOException {

  ContainerManagementProtocolProxyData proxy = null;
  List<ContainerId> containerIds = new ArrayList<ContainerId>();
  containerIds.add(containerId);
  try {
    proxy = cmProxy.getProxy(nodeId.toString(), containerId);
    GetContainerStatusesResponse response =
        proxy.getContainerManagementProtocol().getContainerStatuses(
            GetContainerStatusesRequest.newInstance(containerIds));
    if (response.getFailedRequests() != null
        && response.getFailedRequests().containsKey(containerId)) {
      Throwable t =
          response.getFailedRequests().get(containerId).deSerialize();
      parseAndThrowException(t);
    }
    ContainerStatus containerStatus = response.getContainerStatuses().get(0);
    return containerStatus;
  } finally {
    if (proxy != null) {
      cmProxy.mayBeCloseProxy(proxy);
    }
  }
}

@Public
@Stable
public static AllocateResponse newInstance(int responseId,
    List<ContainerStatus> completedContainers,
    List<Container> allocatedContainers, List<NodeReport> updatedNodes,
    Resource availResources, AMCommand command, int numClusterNodes,
    PreemptionMessage preempt, List<NMToken> nmTokens,
    List<ContainerResourceIncrease> increasedContainers,
    List<ContainerResourceDecrease> decreasedContainers) {
  AllocateResponse response = newInstance(responseId, completedContainers,
      allocatedContainers, updatedNodes, availResources, command,
      numClusterNodes, preempt, nmTokens);
  response.setIncreasedContainers(increasedContainers);
  response.setDecreasedContainers(decreasedContainers);
  return response;
}

@Test
public void testCreateAbnormalContainerStatus() {
  ContainerStatus cd = SchedulerUtils.createAbnormalContainerStatus(
      ContainerId.newContainerId(ApplicationAttemptId.newInstance(
        ApplicationId.newInstance(System.currentTimeMillis(), 1), 1), 1), "x");
  Assert.assertEquals(ContainerExitStatus.ABORTED, cd.getExitStatus());
}

@Override
public void onContainerStatusReceived(ContainerId containerId,
    ContainerStatus containerStatus) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Container Status: id=" + containerId + ", status=" +
        containerStatus);
  }
}

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

@Override
public void setContainerStatuses(List<ContainerStatus> statuses) {
  maybeInitBuilder();
  if (statuses == null)
    builder.clearStatus();
  this.containerStatuses = statuses;
}

@Override
public synchronized void setCompletedContainersStatuses(
    final List<ContainerStatus> containers) {
  if (containers == null)
    return;
  initLocalFinishedContainerList();
  completedContainersStatuses.addAll(containers);
}

private synchronized Iterable<ContainerStatusProto>
getContainerStatusProtoIterable(
    final List<ContainerStatus> newContainersList) {
  maybeInitBuilder();
  return new Iterable<ContainerStatusProto>() {
    @Override
    public synchronized Iterator<ContainerStatusProto> iterator() {
      return new Iterator<ContainerStatusProto>() {

        Iterator<ContainerStatus> iter = newContainersList.iterator();

        @Override
        public synchronized boolean hasNext() {
          return iter.hasNext();
        }

        @Override
        public synchronized ContainerStatusProto next() {
          return convertToProtoFormat(iter.next());
        }

        @Override
        public synchronized void remove() {
          throw new UnsupportedOperationException();

        }
      };

    }
  };
}

static LeafQueue stubLeafQueue(LeafQueue queue) {

  // Mock some methods for ease in these unit tests

  // 1. LeafQueue.createContainer to return dummy containers
  doAnswer(
      new Answer<Container>() {
        @Override
        public Container answer(InvocationOnMock invocation) 
            throws Throwable {
          final FiCaSchedulerApp application = 
              (FiCaSchedulerApp)(invocation.getArguments()[0]);
          final ContainerId containerId =                 
              TestUtils.getMockContainerId(application);

          Container container = TestUtils.getMockContainer(
              containerId,
              ((FiCaSchedulerNode)(invocation.getArguments()[1])).getNodeID(), 
              (Resource)(invocation.getArguments()[2]),
              ((Priority)invocation.getArguments()[3]));
          return container;
        }
      }
    ).
    when(queue).createContainer(
            any(FiCaSchedulerApp.class), 
            any(FiCaSchedulerNode.class), 
            any(Resource.class),
            any(Priority.class)
            );

  // 2. Stub out LeafQueue.parent.completedContainer
  CSQueue parent = queue.getParent();
  doNothing().when(parent).completedContainer(
      any(Resource.class), any(FiCaSchedulerApp.class), any(FiCaSchedulerNode.class), 
      any(RMContainer.class), any(ContainerStatus.class), 
      any(RMContainerEventType.class), any(CSQueue.class), anyBoolean());

  return queue;
}

public static NodeStatus newInstance(NodeId nodeId, int responseId,
    List<ContainerStatus> containerStatuses,
    List<ApplicationId> keepAliveApplications,
    NodeHealthStatus nodeHealthStatus) {
  NodeStatus nodeStatus = Records.newRecord(NodeStatus.class);
  nodeStatus.setResponseId(responseId);
  nodeStatus.setNodeId(nodeId);
  nodeStatus.setContainersStatuses(containerStatuses);
  nodeStatus.setKeepAliveApplications(keepAliveApplications);
  nodeStatus.setNodeHealthStatus(nodeHealthStatus);
  return nodeStatus;
}

private synchronized void addContainersToProto() {
  maybeInitBuilder();
  builder.clearContainersStatuses();
  if (containers == null)
    return;
  Iterable<ContainerStatusProto> iterable = new Iterable<ContainerStatusProto>() {
    @Override
    public Iterator<ContainerStatusProto> iterator() {
      return new Iterator<ContainerStatusProto>() {

        Iterator<ContainerStatus> iter = containers.iterator();

        @Override
        public boolean hasNext() {
          return iter.hasNext();
        }

        @Override
        public ContainerStatusProto next() {
          return convertToProtoFormat(iter.next());
        }

        @Override
        public void remove() {
          throw new UnsupportedOperationException();

        }
      };

    }
  };
  builder.addAllContainersStatuses(iterable);
}

@Override
public synchronized void setContainersStatuses(
    List<ContainerStatus> containers) {
  if (containers == null) {
    builder.clearContainersStatuses();
  }
  this.containers = containers;
}

private synchronized void initContainers() {
  if (this.containers != null) {
    return;
  }
  NodeStatusProtoOrBuilder p = viaProto ? proto : builder;
  List<ContainerStatusProto> list = p.getContainersStatusesList();
  this.containers = new ArrayList<ContainerStatus>();

  for (ContainerStatusProto c : list) {
    this.containers.add(convertFromProtoFormat(c));
  }

}

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

private ContainerStatus getContainerStatus(int applicationId,
    int containerID, int appAttemptId) {
  ContainerStatus status = recordFactory
      .newRecordInstance(ContainerStatus.class);
  status.setContainerId(getContainerId(containerID, appAttemptId));
  return status;
}

public ContainerInfo(final Context nmContext, final Container container,
     String requestUri, String pathPrefix) {

  this.id = container.getContainerId().toString();
  this.nodeId = nmContext.getNodeId().toString();
  ContainerStatus containerData = container.cloneAndGetContainerStatus();
  this.exitCode = containerData.getExitStatus();
  this.exitStatus =
      (this.exitCode == ContainerExitStatus.INVALID) ?
          "N/A" : String.valueOf(exitCode);
  this.state = container.getContainerState().toString();
  this.diagnostics = containerData.getDiagnostics();
  if (this.diagnostics == null || this.diagnostics.isEmpty()) {
    this.diagnostics = "";
  }

  this.user = container.getUser();
  Resource res = container.getResource();
  if (res != null) {
    this.totalMemoryNeededMB = res.getMemory();
    this.totalVCoresNeeded = res.getVirtualCores();
  }
  this.containerLogsShortLink = ujoin("containerlogs", this.id,
      container.getUser());

  if (requestUri == null) {
    requestUri = "";
  }
  if (pathPrefix == null) {
    pathPrefix = "";
  }
  this.containerLogsLink = join(requestUri, pathPrefix,
      this.containerLogsShortLink);
}

@Override
public ContainerStatus cloneAndGetContainerStatus() {
  this.readLock.lock();
  try {
    return BuilderUtils.newContainerStatus(this.containerId,
      getCurrentState(), diagnostics.toString(), exitCode);
  } finally {
    this.readLock.unlock();
  }
}

@Test (timeout = 5000)
public void testCallAMRMClientAsyncStopFromCallbackHandler()
    throws YarnException, IOException, InterruptedException {
  Configuration conf = new Configuration();
  TestCallbackHandler2 callbackHandler = new TestCallbackHandler2();
  @SuppressWarnings("unchecked")
  AMRMClient<ContainerRequest> client = mock(AMRMClientImpl.class);

  List<ContainerStatus> completed = Arrays.asList(
      ContainerStatus.newInstance(newContainerId(0, 0, 0, 0),
          ContainerState.COMPLETE, "", 0));
  final AllocateResponse response = createAllocateResponse(completed,
      new ArrayList<Container>(), null);

  when(client.allocate(anyFloat())).thenReturn(response);

  AMRMClientAsync<ContainerRequest> asyncClient =
      AMRMClientAsync.createAMRMClientAsync(client, 20, callbackHandler);
  callbackHandler.asynClient = asyncClient;
  asyncClient.init(conf);
  asyncClient.start();

  synchronized (callbackHandler.notifier) {
    asyncClient.registerApplicationMaster("localhost", 1234, null);
    while(callbackHandler.notify == false) {
      try {
        callbackHandler.notifier.wait();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
}

@Override
public ContainerStatus cloneAndGetContainerStatus() {
  ContainerStatus containerStatus = recordFactory
      .newRecordInstance(ContainerStatus.class);
  containerStatus
      .setState(org.apache.hadoop.yarn.api.records.ContainerState.RUNNING);
  containerStatus.setDiagnostics("testing");
  containerStatus.setExitStatus(0);
  return containerStatus;
}

public NodeHeartbeatResponse nodeHeartbeat(Map<ApplicationId,
    List<ContainerStatus>> conts, boolean isHealthy, int resId) throws Exception {
  NodeHeartbeatRequest req = Records.newRecord(NodeHeartbeatRequest.class);
  NodeStatus status = Records.newRecord(NodeStatus.class);
  status.setResponseId(resId);
  status.setNodeId(nodeId);
  for (Map.Entry<ApplicationId, List<ContainerStatus>> entry : conts.entrySet()) {
    Log.info("entry.getValue() " + entry.getValue());
    status.setContainersStatuses(entry.getValue());
  }
  NodeHealthStatus healthStatus = Records.newRecord(NodeHealthStatus.class);
  healthStatus.setHealthReport("");
  healthStatus.setIsNodeHealthy(isHealthy);
  healthStatus.setLastHealthReportTime(1);
  status.setNodeHealthStatus(healthStatus);
  req.setNodeStatus(status);
  req.setLastKnownContainerTokenMasterKey(this.currentContainerTokenMasterKey);
  req.setLastKnownNMTokenMasterKey(this.currentNMTokenMasterKey);
  NodeHeartbeatResponse heartbeatResponse =
      resourceTracker.nodeHeartbeat(req);

  MasterKey masterKeyFromRM = heartbeatResponse.getContainerTokenMasterKey();
  if (masterKeyFromRM != null
      && masterKeyFromRM.getKeyId() != this.currentContainerTokenMasterKey
          .getKeyId()) {
    this.currentContainerTokenMasterKey = masterKeyFromRM;
  }

  masterKeyFromRM = heartbeatResponse.getNMTokenMasterKey();
  if (masterKeyFromRM != null
      && masterKeyFromRM.getKeyId() != this.currentNMTokenMasterKey
          .getKeyId()) {
    this.currentNMTokenMasterKey = masterKeyFromRM;
  }

  return heartbeatResponse;
}

/**
 * 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);
  }
}

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

@Override
public GetContainerStatusesResponse getContainerStatuses(
    GetContainerStatusesRequest request) throws IOException {
  List<ContainerStatus> statuses = new ArrayList<ContainerStatus>();
  statuses.add(status);
  return GetContainerStatusesResponse.newInstance(statuses, null);
}

private void handleNMContainerStatus(
    List<NMContainerStatus> nmContainerStatuses, RMNodeImpl rmnode) {
  List<ContainerStatus> containerStatuses =
      new ArrayList<ContainerStatus>();
  for (NMContainerStatus nmContainerStatus : nmContainerStatuses) {
    containerStatuses.add(createContainerStatus(nmContainerStatus));
  }
  rmnode.handleContainerStatus(containerStatuses);
}