import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
public synchronized void recoverContainer(SchedulerNode node,
    RMContainer rmContainer) {
  // recover app scheduling info
  appSchedulingInfo.recoverContainer(rmContainer);

  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  LOG.info("SchedulerAttempt " + getApplicationAttemptId()
    + " is recovering container " + rmContainer.getContainerId());
  liveContainers.put(rmContainer.getContainerId(), rmContainer);
  attemptResourceUsage.incUsed(node.getPartition(), rmContainer
      .getContainer().getResource());
  
  // resourceLimit: updated when LeafQueue#recoverContainer#allocateResource
  // is called.
  // newlyAllocatedContainers.add(rmContainer);
  // schedulingOpportunities
  // lastScheduledContainer
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
public synchronized void recoverContainer(RMContainer rmContainer) {
  QueueMetrics metrics = queue.getMetrics();
  if (pending) {
    // If there was any container to recover, the application was
    // running from scheduler's POV.
    pending = false;
    metrics.runAppAttempt(applicationId, user);
  }

  // Container is completed. Skip recovering resources.
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }

  metrics.allocateResources(user, 1, rmContainer.getAllocatedResource(),
    false);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  // Careful! Locking order is important! 
  synchronized (this) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.allocateResource(clusterResource, rmContainer.getContainer()
        .getResource(), node.getLabels());
  }
  if (parent != null) {
    parent.recoverContainer(clusterResource, attempt, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
private LogAggregationContext getLogAggregationContextFromContainerToken(
    MockRM rm1, MockNM nm1, LogAggregationContext logAggregationContext)
    throws Exception {
  RMApp app2 = rm1.submitApp(200, logAggregationContext);
  MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);
  nm1.nodeHeartbeat(true);
  // request a container.
  am2.allocate("127.0.0.1", 512, 1, new ArrayList<ContainerId>());
  ContainerId containerId =
      ContainerId.newContainerId(am2.getApplicationAttemptId(), 2);
  rm1.waitForState(nm1, containerId, RMContainerState.ALLOCATED);

  // acquire the container.
  List<Container> containers =
      am2.allocate(new ArrayList<ResourceRequest>(),
        new ArrayList<ContainerId>()).getAllocatedContainers();
  Assert.assertEquals(containerId, containers.get(0).getId());
  // container token is generated.
  Assert.assertNotNull(containers.get(0).getContainerToken());
  ContainerTokenIdentifier token =
      BuilderUtils.newContainerTokenIdentifier(containers.get(0)
        .getContainerToken());
  return token.getLogAggregationContext();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  // Careful! Locking order is important! 
  synchronized (this) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.allocateResource(clusterResource, rmContainer.getContainer()
        .getResource(), node.getPartition(), false);
  }
  if (parent != null) {
    parent.recoverContainer(clusterResource, attempt, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
private void allocateAndLaunchContainers(MockAM am, MockNM nm, MockRM rm,
    int nContainer, int mem, int priority, int startContainerId)
        throws Exception {
  am.allocate(Arrays
      .asList(ResourceRequest.newInstance(Priority.newInstance(priority), "*",
          Resources.createResource(mem), nContainer)),
      null);
  ContainerId lastContainerId = ContainerId.newContainerId(
      am.getApplicationAttemptId(), startContainerId + nContainer - 1);
  Assert.assertTrue(rm.waitForState(nm, lastContainerId,
      RMContainerState.ALLOCATED, 10 * 1000));
  // Acquire them, and NM report RUNNING
  am.allocate(null, null);

  for (int cId = startContainerId; cId < startContainerId
      + nContainer; cId++) {
    sentRMContainerLaunched(rm,
        ContainerId.newContainerId(am.getApplicationAttemptId(), cId));
    rm.waitForContainerState(
        ContainerId.newContainerId(am.getApplicationAttemptId(), cId),
        RMContainerState.RUNNING);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
public synchronized void recoverContainer(RMContainer rmContainer) {
  // recover app scheduling info
  appSchedulingInfo.recoverContainer(rmContainer);

  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  LOG.info("SchedulerAttempt " + getApplicationAttemptId()
    + " is recovering container " + rmContainer.getContainerId());
  liveContainers.put(rmContainer.getContainerId(), rmContainer);
  Resources.addTo(currentConsumption, rmContainer.getContainer()
    .getResource());
  // resourceLimit: updated when LeafQueue#recoverContainer#allocateResource
  // is called.
  // newlyAllocatedContainers.add(rmContainer);
  // schedulingOpportunities
  // lastScheduledContainer
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  // Careful! Locking order is important! 
  synchronized (this) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.allocateResource(clusterResource, rmContainer.getContainer()
        .getResource(), node.getLabels(),false);
  }
  if (parent != null) {
    parent.recoverContainer(clusterResource, attempt, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  // Careful! Locking order is important! 
  synchronized (this) {
    super.allocateResource(clusterResource, rmContainer.getContainer()
        .getResource(), labelManager.getLabelsOnNode(rmContainer
        .getContainer().getNodeId()));
  }
  if (parent != null) {
    parent.recoverContainer(clusterResource, attempt, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  // Careful! Locking order is important! 
  synchronized (this) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    allocateResource(clusterResource,
        rmContainer.getContainer().getResource(), node.getPartition(), false);
  }
  if (parent != null) {
    parent.recoverContainer(clusterResource, attempt, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
protected void decreaseContainer(SchedContainerChangeRequest decreaseRequest,
    SchedulerApplicationAttempt attempt) {
  RMContainer rmContainer = decreaseRequest.getRMContainer();
  // Check container status before doing decrease
  if (rmContainer.getState() != RMContainerState.RUNNING) {
    LOG.info("Trying to decrease a container not in RUNNING state, container="
        + rmContainer + " state=" + rmContainer.getState().name());
    return;
  }
  FiCaSchedulerApp app = (FiCaSchedulerApp)attempt;
  LeafQueue queue = (LeafQueue) attempt.getQueue();
  try {
    queue.decreaseContainer(clusterResource, decreaseRequest, app);
    // Notify RMNode that the container can be pulled by NodeManager in the
    // next heartbeat
    this.rmContext.getDispatcher().getEventHandler()
        .handle(new RMNodeDecreaseContainerEvent(
            decreaseRequest.getNodeId(),
            Collections.singletonList(rmContainer.getContainer())));
  } catch (InvalidResourceRequestException e) {
    LOG.warn("Error happens when checking decrease request, Ignoring.."
        + " exception=", e);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
public void waitForState(MockNM nm, ContainerId containerId,
    RMContainerState containerState) throws Exception {
  RMContainer container = getResourceScheduler().getRMContainer(containerId);
  int timeoutSecs = 0;
  while(container == null && timeoutSecs++ < 20) {
    nm.nodeHeartbeat(true);
    container = getResourceScheduler().getRMContainer(containerId);
    System.out.println("Waiting for container " + containerId + " to be allocated.");
    Thread.sleep(100);
  }
  Assert.assertNotNull("Container shouldn't be null", container);
  timeoutSecs = 0;
  while (!containerState.equals(container.getState()) && timeoutSecs++ < 40) {
    System.out.println("Container : " + containerId + " State is : "
        + container.getState() + " Waiting for state : " + containerState);
    nm.nodeHeartbeat(true);
    Thread.sleep(300);
  }
  System.out.println("Container State is : " + container.getState());
  Assert.assertEquals("Container state is not correct (timedout)",
    containerState, container.getState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt schedulerAttempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  increaseUsedResources(rmContainer);
  updateAppHeadRoom(schedulerAttempt);
  updateAvailableResourcesMetrics();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
private synchronized void doneApplicationAttempt(
    ApplicationAttemptId applicationAttemptId,
    RMAppAttemptState rmAppAttemptFinalState, boolean keepContainers)
    throws IOException {
  FiCaSchedulerApp attempt = getApplicationAttempt(applicationAttemptId);
  SchedulerApplication<FiCaSchedulerApp> application =
      applications.get(applicationAttemptId.getApplicationId());
  if (application == null || attempt == null) {
    throw new IOException("Unknown application " + applicationAttemptId + 
    " has completed!");
  }

  // Kill all 'live' containers
  for (RMContainer container : attempt.getLiveContainers()) {
    if (keepContainers
        && container.getState().equals(RMContainerState.RUNNING)) {
      // do not kill the running container in the case of work-preserving AM
      // restart.
      LOG.info("Skip killing " + container.getContainerId());
      continue;
    }
    completedContainer(container,
      SchedulerUtils.createAbnormalContainerStatus(
        container.getContainerId(), SchedulerUtils.COMPLETED_APPLICATION),
      RMContainerEventType.KILL);
  }

  // Clean up pending requests, metrics etc.
  attempt.stop(rmAppAttemptFinalState);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
/**
 * Clean up a completed container.
 */
@Override
protected synchronized void completedContainer(RMContainer rmContainer,
    ContainerStatus containerStatus, RMContainerEventType event) {
  if (rmContainer == null) {
    LOG.info("Null container completed...");
    return;
  }

  Container container = rmContainer.getContainer();

  // Get the application for the finished container
  FSAppAttempt application =
      getCurrentAttemptForContainer(container.getId());
  ApplicationId appId =
      container.getId().getApplicationAttemptId().getApplicationId();
  if (application == null) {
    LOG.info("Container " + container + " of" +
        " unknown application attempt " + appId +
        " completed with event " + event);
    return;
  }

  // Get the node on which the container was allocated
  FSSchedulerNode node = getFSSchedulerNode(container.getNodeId());

  if (rmContainer.getState() == RMContainerState.RESERVED) {
    application.unreserve(rmContainer.getReservedPriority(), node);
  } else {
    application.containerCompleted(rmContainer, containerStatus, event);
    node.releaseContainer(container);
    updateRootQueueMetrics();
  }

  LOG.info("Application attempt " + application.getApplicationAttemptId()
      + " released container " + container.getId() + " on node: " + node
      + " with event: " + event);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Override
public void recoverContainer(Resource clusterResource,
    SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
  if (rmContainer.getState().equals(RMContainerState.COMPLETED)) {
    return;
  }
  // Careful! Locking order is important! 
  synchronized (this) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    allocateResource(clusterResource, attempt, rmContainer.getContainer()
        .getResource(), node.getLabels());
  }
  getParent().recoverContainer(clusterResource, attempt, rmContainer);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
public boolean waitForState(MockNM nm, ContainerId containerId,
    RMContainerState containerState, int timeoutMillisecs) throws Exception {
  RMContainer container = getResourceScheduler().getRMContainer(containerId);
  int timeoutSecs = 0;
  while(container == null && timeoutSecs++ < timeoutMillisecs / 100) {
    nm.nodeHeartbeat(true);
    container = getResourceScheduler().getRMContainer(containerId);
    System.out.println("Waiting for container " + containerId + " to be allocated.");
    Thread.sleep(100);
    
    if (timeoutMillisecs <= timeoutSecs * 100) {
      return false;
    }
  }
  Assert.assertNotNull("Container shouldn't be null", container);
  while (!containerState.equals(container.getState())
      && timeoutSecs++ < timeoutMillisecs / 100) {
    System.out.println("Container : " + containerId + " State is : "
        + container.getState() + " Waiting for state : " + containerState);
    nm.nodeHeartbeat(true);
    Thread.sleep(100);
    
    if (timeoutMillisecs <= timeoutSecs * 100) {
      return false;
    }
  }
  
  System.out.println("Container State is : " + container.getState());
  Assert.assertEquals("Container state is not correct (timedout)",
    containerState, container.getState());
  return true;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
private void waitContainerAllocated(MockAM am, int mem, int nContainer,
    int startContainerId, MockRM rm, MockNM nm) throws Exception {
  for (int cId = startContainerId; cId < startContainerId + nContainer; cId++) {
    am.allocate("*", mem, 1, new ArrayList<ContainerId>());
    ContainerId containerId =
        ContainerId.newContainerId(am.getApplicationAttemptId(), cId);
    Assert.assertTrue(rm.waitForState(nm, containerId,
        RMContainerState.ALLOCATED, 10 * 1000));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Test
public void testContainerTokenGeneratedOnPullRequest() throws Exception {
  MockRM rm1 = new MockRM(conf);
  rm1.start();
  MockNM nm1 = rm1.registerNode("127.0.0.1:1234", 8000);
  RMApp app1 = rm1.submitApp(200);
  MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
  // request a container.
  am1.allocate("127.0.0.1", 1024, 1, new ArrayList<ContainerId>());
  ContainerId containerId2 =
      ContainerId.newContainerId(am1.getApplicationAttemptId(), 2);
  rm1.waitForState(nm1, containerId2, RMContainerState.ALLOCATED);

  RMContainer container =
      rm1.getResourceScheduler().getRMContainer(containerId2);
  // no container token is generated.
  Assert.assertEquals(containerId2, container.getContainerId());
  Assert.assertNull(container.getContainer().getContainerToken());

  // acquire the container.
  List<Container> containers =
      am1.allocate(new ArrayList<ResourceRequest>(),
        new ArrayList<ContainerId>()).getAllocatedContainers();
  Assert.assertEquals(containerId2, containers.get(0).getId());
  // container token is generated.
  Assert.assertNotNull(containers.get(0).getContainerToken());
  rm1.stop();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; //导入依赖的package包/类
@Test
public void testNormalContainerAllocationWhenDNSUnavailable() throws Exception{
  MockRM rm1 = new MockRM(conf);
  rm1.start();
  MockNM nm1 = rm1.registerNode("unknownhost:1234", 8000);
  RMApp app1 = rm1.submitApp(200);
  MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);

  // request a container.
  am1.allocate("127.0.0.1", 1024, 1, new ArrayList<ContainerId>());
  ContainerId containerId2 =
      ContainerId.newContainerId(am1.getApplicationAttemptId(), 2);
  rm1.waitForState(nm1, containerId2, RMContainerState.ALLOCATED);

  // acquire the container.
  SecurityUtilTestHelper.setTokenServiceUseIp(true);
  List<Container> containers =
      am1.allocate(new ArrayList<ResourceRequest>(),
        new ArrayList<ContainerId>()).getAllocatedContainers();
  // not able to fetch the container;
  Assert.assertEquals(0, containers.size());

  SecurityUtilTestHelper.setTokenServiceUseIp(false);
  containers =
      am1.allocate(new ArrayList<ResourceRequest>(),
        new ArrayList<ContainerId>()).getAllocatedContainers();
  // should be able to fetch the container;
  Assert.assertEquals(1, containers.size());
}