import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@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);
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
public MockRMNodeImpl(NodeId nodeId, String nodeAddr, String httpAddress,
    Resource perNode, String rackName, String healthReport,
    long lastHealthReportTime, int cmdPort, String hostName, NodeState state,
    Set<String> labels) {
  this.nodeId = nodeId;
  this.nodeAddr = nodeAddr;
  this.httpAddress = httpAddress;
  this.perNode = perNode;
  this.rackName = rackName;
  this.healthReport = healthReport;
  this.lastHealthReportTime = lastHealthReportTime;
  this.cmdPort = cmdPort;
  this.hostName = hostName;
  this.state = state;
  this.labels = labels;
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private void initLabelsToNodes() {
  if (this.labelsToNodes != null) {
    return;
  }
  GetLabelsToNodesResponseProtoOrBuilder p = viaProto ? proto : builder;
  List<LabelsToNodeIdsProto> list = p.getLabelsToNodesList();
  this.labelsToNodes = new HashMap<String, Set<NodeId>>();

  for (LabelsToNodeIdsProto c : list) {
    Set<NodeId> setNodes = new HashSet<NodeId>();
    for(NodeIdProto n : c.getNodeIdList()) {
      NodeId node = new NodeIdPBImpl(n);
      setNodes.add(node);
    }
    if(!setNodes.isEmpty()) {
      this.labelsToNodes.put(c.getNodeLabels(), setNodes);
    }
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private static void uploadContainerLogIntoRemoteDir(UserGroupInformation ugi,
    Configuration configuration, List<String> rootLogDirs, NodeId nodeId,
    ContainerId containerId, Path appDir, FileSystem fs) throws Exception {
  Path path =
      new Path(appDir, LogAggregationUtils.getNodeString(nodeId)
          + System.currentTimeMillis());
  AggregatedLogFormat.LogWriter writer =
      new AggregatedLogFormat.LogWriter(configuration, path, ugi);
  writer.writeApplicationOwner(ugi.getUserName());

  Map<ApplicationAccessType, String> appAcls =
      new HashMap<ApplicationAccessType, String>();
  appAcls.put(ApplicationAccessType.VIEW_APP, ugi.getUserName());
  writer.writeApplicationACLs(appAcls);
  writer.append(new AggregatedLogFormat.LogKey(containerId),
    new AggregatedLogFormat.LogValue(rootLogDirs, containerId,
      UserGroupInformation.getCurrentUser().getShortUserName()));
  writer.close();
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Test
public void testLogsView2() throws IOException {
  LOG.info("HsLogsPage with data");
  MockAppContext ctx = new MockAppContext(0, 1, 1, 1);
  Map<String, String> params = new HashMap<String, String>();

  params.put(CONTAINER_ID, MRApp.newContainerId(1, 1, 333, 1)
      .toString());
  params.put(NM_NODENAME, 
      NodeId.newInstance(MockJobs.NM_HOST, MockJobs.NM_PORT).toString());
  params.put(ENTITY_STRING, "container_10_0001_01_000001");
  params.put(APP_OWNER, "owner");

  Injector injector =
      WebAppTests.testPage(AggregatedLogsPage.class, AppContext.class, ctx,
          params);
  PrintWriter spyPw = WebAppTests.getPrintWriter(injector);
  verify(spyPw).write(
      "Aggregation is not enabled. Try the nodemanager at "
          + MockJobs.NM_HOST + ":" + MockJobs.NM_PORT);
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private void computeRackAndLocality() {
  NodeId containerNodeId = container.getNodeId();
  nodeRackName = RackResolver.resolve(
      containerNodeId.getHost()).getNetworkLocation();

  locality = Locality.OFF_SWITCH;
  if (dataLocalHosts.size() > 0) {
    String cHost = resolveHost(containerNodeId.getHost());
    if (dataLocalHosts.contains(cHost)) {
      locality = Locality.NODE_LOCAL;
    }
  }
  if (locality == Locality.OFF_SWITCH) {
    if (dataLocalRacks.contains(nodeRackName)) {
      locality = Locality.RACK_LOCAL;
    }
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Override
public void addLabelsToNode(Map<NodeId, Set<String>> addedLabelsToNode)
    throws IOException {    
  try {
    writeLock.lock();

    // get nodesCollection before edition
    Map<String, Host> before = cloneNodeMap(addedLabelsToNode.keySet());

    super.addLabelsToNode(addedLabelsToNode);

    // get nodesCollection after edition
    Map<String, Host> after = cloneNodeMap(addedLabelsToNode.keySet());

    // update running nodes resources
    updateResourceMappings(before, after);
  } finally {
    writeLock.unlock();
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private String testGetContainer(YarnRPC rpc,
    ApplicationAttemptId appAttemptId, NodeId nodeId,
    ContainerId containerId,
    org.apache.hadoop.yarn.api.records.Token nmToken,
    boolean isExceptionExpected) {
  try {
    getContainerStatus(rpc, nmToken, containerId, appAttemptId, nodeId,
        isExceptionExpected);
    if (isExceptionExpected) {
      fail("Exception was expected!!");
    }
    return "";
  } catch (Exception e) {
    e.printStackTrace();
    return e.getMessage();
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
public void transition(RMAppImpl app, RMAppEvent event) {
  for (NodeId nodeId : app.getRanNodes()) {
    app.handler.handle(
        new RMNodeCleanAppEvent(nodeId, app.applicationId));
  }
  app.finishTime = app.storedFinishTime;
  if (app.finishTime == 0 ) {
    app.finishTime = app.systemClock.getTime();
  }
  // Recovered apps that are completed were not added to scheduler, so no
  // need to remove them from scheduler.
  if (app.recoveredFinalState == null) {
    app.handler.handle(new AppRemovedSchedulerEvent(app.applicationId,
      finalState));
  }
  app.handler.handle(
      new RMAppManagerEvent(app.applicationId,
      RMAppManagerEventType.APP_COMPLETED));

  app.rmContext.getRMApplicationHistoryWriter()
      .applicationFinished(app, finalState);
  app.rmContext.getSystemMetricsPublisher()
      .appFinished(app, finalState, app.finishTime);
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Test
public void testAbsentNodeStatus() throws Exception {
  NodeId nodeId = NodeId.newInstance("Absenthost0", 0);
  NodeCLI cli = new NodeCLI();
  when(client.getNodeReports()).thenReturn(
              getNodeReports(0, NodeState.RUNNING));
  cli.setClient(client);
  cli.setSysOutPrintStream(sysOut);
  cli.setSysErrPrintStream(sysErr);
  int result = cli.run(new String[] { "-status", nodeId.toString() });
  assertEquals(0, result);
  verify(client).getNodeReports();
  verify(sysOut, times(1)).println(isA(String.class));
  verify(sysOut).println(
    "Could not find the node report for node id : " + nodeId.toString());
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Override
protected RMContainerTokenSecretManager createContainerTokenSecretManager(
    Configuration conf) {
  return new RMContainerTokenSecretManager(conf) {

    @Override
    public Token createContainerToken(ContainerId containerId,
        NodeId nodeId, String appSubmitter, Resource capability,
        Priority priority, long createTime,
        LogAggregationContext logAggregationContext) {
      numRetries++;
      return super.createContainerToken(containerId, nodeId, appSubmitter,
        capability, priority, createTime, logAggregationContext);
    }
  };
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Test
public void testNodeRegistrationFailure() throws Exception {
  writeToHostsFile("host1");
  Configuration conf = new Configuration();
  conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile
      .getAbsolutePath());
  rm = new MockRM(conf);
  rm.start();
  
  ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService();
  RegisterNodeManagerRequest req = Records.newRecord(
      RegisterNodeManagerRequest.class);
  NodeId nodeId = NodeId.newInstance("host2", 1234);
  req.setNodeId(nodeId);
  req.setHttpPort(1234);
  // trying to register a invalid node.
  RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req);
  Assert.assertEquals(NodeAction.SHUTDOWN,response.getNodeAction());
  Assert
    .assertEquals(
      "Disallowed NodeManager from  host2, Sending SHUTDOWN signal to the NodeManager.",
      response.getDiagnosticsMessage());
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private String testStopContainer(YarnRPC rpc,
    ApplicationAttemptId appAttemptId, NodeId nodeId,
    ContainerId containerId, Token nmToken, boolean isExceptionExpected) {
  try {
    stopContainer(rpc, nmToken,
        Arrays.asList(new ContainerId[] { containerId }), appAttemptId,
        nodeId);
    if (isExceptionExpected) {
      fail("Exception was expected!!");
    }
    return "";
  } catch (Exception e) {
    e.printStackTrace();
    return e.getMessage();
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Private
@VisibleForTesting
public boolean isApplicationAttemptNMTokenPresent(
    ApplicationAttemptId appAttemptId, NodeId nodeId) {
  try {
    this.readLock.lock();
    HashSet<NodeId> nodes = this.appAttemptToNodeKeyMap.get(appAttemptId);
    if (nodes != null && nodes.contains(nodeId)) {
      return true;
    } else {
      return false;
    }
  } finally {
    this.readLock.unlock();
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private boolean waitForNodeManagerToConnect(int timeout, NodeId nodeId)
    throws Exception {
  for (int i = 0; i < timeout / 100; i++) {
    if (getActiveRM().getRMContext().getRMNodes().containsKey(nodeId)) {
      return true;
    }
    Thread.sleep(100);
  }
  return false;
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void handleUpdatedNodes(AllocateResponse response) {
  // send event to the job about on updated nodes
  List<NodeReport> updatedNodes = response.getUpdatedNodes();
  if (!updatedNodes.isEmpty()) {

    // send event to the job to act upon completed tasks
    eventHandler.handle(new JobUpdatedNodesEvent(getJob().getID(),
        updatedNodes));

    // act upon running tasks
    HashSet<NodeId> unusableNodes = new HashSet<NodeId>();
    for (NodeReport nr : updatedNodes) {
      NodeState nodeState = nr.getNodeState();
      if (nodeState.isUnusable()) {
        unusableNodes.add(nr.getNodeId());
      }
    }
    for (int i = 0; i < 2; ++i) {
      HashMap<TaskAttemptId, Container> taskSet = i == 0 ? assignedRequests.maps
          : assignedRequests.reduces;
      // kill running containers
      for (Map.Entry<TaskAttemptId, Container> entry : taskSet.entrySet()) {
        TaskAttemptId tid = entry.getKey();
        NodeId taskAttemptNodeId = entry.getValue().getNodeId();
        if (unusableNodes.contains(taskAttemptNodeId)) {
          LOG.info("Killing taskAttempt:" + tid
              + " because it is running on unusable node:"
              + taskAttemptNodeId);
          eventHandler.handle(new TaskAttemptKillEvent(tid,
              "TaskAttempt killed because it ran on unusable node"
                  + taskAttemptNodeId));
        }
      }
    }
  }
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private RMNodeImpl getRebootedNode() {
  NodeId nodeId = BuilderUtils.newNodeId("localhost", 0);
  Resource capability = Resource.newInstance(4096, 4, 4);
  RMNodeImpl node = new RMNodeImpl(nodeId, rmContext,null, 0, 0,
      null, capability, null);
  node.handle(new RMNodeStartedEvent(node.getNodeID(), null, null));
  Assert.assertEquals(NodeState.RUNNING, node.getState());
  node.handle(new RMNodeEvent(node.getNodeID(), RMNodeEventType.REBOOTING));
  Assert.assertEquals(NodeState.REBOOTED, node.getState());
  return node;
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
private void replaceNodeForLabels(NodeId node, Set<String> oldLabels,
    Set<String> newLabels) {
  if(oldLabels != null) {
    removeNodeFromLabels(node, oldLabels);
  }
  addNodeToLabels(node, newLabels);
}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@Override
protected ContainerLauncher
    createContainerLauncher(final AppContext context) {
  return new ContainerLauncherImpl(context) {

    @Override
    public ContainerManagementProtocolProxyData getCMProxy(
        String containerMgrBindAddr, ContainerId containerId)
        throws IOException {
      InetSocketAddress addr = NetUtils.getConnectAddress(server);
      String containerManagerBindAddr =
          addr.getHostName() + ":" + addr.getPort();
      Token token =
          tokenSecretManager.createNMToken(
            containerId.getApplicationAttemptId(),
            NodeId.newInstance(addr.getHostName(), addr.getPort()), "user");
      ContainerManagementProtocolProxy cmProxy =
          new ContainerManagementProtocolProxy(conf);
      ContainerManagementProtocolProxyData proxy =
          cmProxy.new ContainerManagementProtocolProxyData(
            YarnRPC.create(conf), containerManagerBindAddr, containerId,
            token);
      return proxy;
    }
  };

}
 

import org.apache.hadoop.yarn.api.records.NodeId; //导入依赖的package包/类
@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));
}