import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getRunningContainers()) {
    completedContainer(container, 
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  updateMaximumAllocation(node, false);
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getRMNode().getTotalCapability());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
protected void releaseContainers(List<ContainerId> containers,
    SchedulerApplicationAttempt attempt) {
  for (ContainerId containerId : containers) {
    RMContainer rmContainer = getRMContainer(containerId);
    if (rmContainer == null) {
      if (System.currentTimeMillis() - ResourceManager.getClusterTimeStamp()
          < nmExpireInterval) {
        LOG.info(containerId + " doesn't exist. Add the container"
            + " to the release request cache as it maybe on recovery.");
        synchronized (attempt) {
          attempt.getPendingRelease().add(containerId);
        }
      } else {
        RMAuditLogger.logFailure(attempt.getUser(),
          AuditConstants.RELEASE_CONTAINER,
          "Unauthorized access or invalid container", "Scheduler",
          "Trying to release container not owned by app or with invalid id.",
          attempt.getApplicationId(), containerId);
      }
    }
    completedContainer(rmContainer,
      SchedulerUtils.createAbnormalContainerStatus(containerId,
        SchedulerUtils.RELEASED_CONTAINER), RMContainerEventType.RELEASED);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getRunningContainers()) {
    completedContainer(container, 
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  updateMaximumAllocation(node, false);
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getTotalResource());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
@Override
public void completedContainer(Resource clusterResource,
    FiCaSchedulerApp application, FiCaSchedulerNode node, 
    RMContainer rmContainer, ContainerStatus containerStatus, 
    RMContainerEventType event, CSQueue completedChildQueue,
    boolean sortQueues) {
  if (application != null) {
    internalReleaseResource(clusterResource, node,
        rmContainer.getContainer().getResource(), false, completedChildQueue,
        sortQueues);

    // Inform the parent
    if (parent != null) {
      // complete my parent
      parent.completedContainer(clusterResource, application, 
          node, rmContainer, null, event, this, sortQueues);
    }    
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
private void handleExcessReservedContainer(Resource clusterResource,
    CSAssignment assignment, FiCaSchedulerNode node, FiCaSchedulerApp app) {
  if (assignment.getExcessReservation() != null) {
    RMContainer excessReservedContainer = assignment.getExcessReservation();
    
    if (excessReservedContainer.hasIncreaseReservation()) {
      unreserveIncreasedContainer(clusterResource,
          app, node, excessReservedContainer);
    } else {
      completedContainer(clusterResource, assignment.getApplication(),
          scheduler.getNode(excessReservedContainer.getAllocatedNode()),
          excessReservedContainer,
          SchedulerUtils.createAbnormalContainerStatus(
              excessReservedContainer.getContainerId(),
              SchedulerUtils.UNRESERVED_CONTAINER),
          RMContainerEventType.RELEASED, null, false);
    }

    assignment.setExcessReservation(null);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
public synchronized void adjusctContianerResouce(ContainerId contianerId, Resource resource){
 
 RMContainer container = getRMContainer(contianerId);
 //get original resource
 Resource allocatedResource = container.getAllocatedResource();
 Resource marginalResource=Resources.subtract(allocatedResource, resource);
 
 if(marginalResource.getMemory() > 0 ){		  
  LOG.info("increase memory on continaer"+contianerId.getContainerId()+"by"+marginalResource.getMemory());		  
 }else{		  
  LOG.info("decrease memory on continaer"+contianerId.getContainerId()+"by"+marginalResource.getMemory());
 }
 
 if(marginalResource.getVirtualCores() > 0){
  LOG.info("increase cores on contianer"+contianerId.getContainerId()+"by"+marginalResource.getVirtualCores());
 }else{
  LOG.info("dncrease cores on contianer"+contianerId.getContainerId()+"by"+marginalResource.getVirtualCores());
 }
 
 SchedulerNode schedulerNode = getSchedulerNode(container.getAllocatedNode());
 //now inform the schedulerNode to adjust resurce assumption
 schedulerNode.addAvailableResource(marginalResource);
 //now inform RMContinaer to adjuct its allocatedResource
 container.handle(new RMContainerResourceUpdateEvent(contianerId, RMContainerEventType.RESOURCE_UPDATE, resource));
 	  
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getRunningContainers()) {
    containerCompleted(container, 
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getRMNode().getTotalCapability());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public synchronized void containerLaunchedOnNode(ContainerId containerId,
    NodeId nodeId) {
  // Inform the container
  RMContainer rmContainer = 
      getRMContainer(containerId);
  if (rmContainer == null) {
    // Some unknown container sneaked into the system. Kill it.
    rmContext.getDispatcher().getEventHandler()
      .handle(new RMNodeCleanContainerEvent(nodeId, containerId));
    return;
  }

  rmContainer.handle(new RMContainerEvent(containerId,
    RMContainerEventType.LAUNCHED));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getCopiedListOfRunningContainers()) {
    super.completedContainer(container,
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  updateMaximumAllocation(node, false);
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getTotalResource());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
@VisibleForTesting
@Private
// clean up a completed container
public void completedContainer(RMContainer rmContainer,
    ContainerStatus containerStatus, RMContainerEventType event) {

  if (rmContainer == null) {
    LOG.info("Container " + containerStatus.getContainerId()
        + " completed with event " + event
        + ", but corresponding RMContainer doesn't exist.");
    return;
  }

  completedContainerInternal(rmContainer, containerStatus, event);

  // If the container is getting killed in ACQUIRED state, the requester (AM
  // for regular containers and RM itself for AM container) will not know what
  // happened. Simply add the ResourceRequest back again so that requester
  // doesn't need to do anything conditionally.
  recoverResourceRequestForContainer(rmContainer);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
@Override
protected void completedContainerInternal(
    RMContainer rmContainer, ContainerStatus containerStatus,
    RMContainerEventType event) {
  Container container = rmContainer.getContainer();
  ContainerId containerId = container.getId();
  
  // Get the application for the finished container
  FiCaSchedulerApp application =
      getCurrentAttemptForContainer(container.getId());
  ApplicationId appId =
      containerId.getApplicationAttemptId().getApplicationId();
  if (application == null) {
    LOG.info("Container " + container + " of" + " finished application "
        + appId + " completed with event " + event);
    return;
  }
  
  // Get the node on which the container was allocated
  FiCaSchedulerNode node = getNode(container.getNodeId());
  
  // Inform the queue
  LeafQueue queue = (LeafQueue)application.getQueue();
  queue.completedContainer(clusterResource, application, node, 
      rmContainer, containerStatus, event, null, true);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
private void killToPreemptContainers(Resource clusterResource,
    FiCaSchedulerNode node,
    CSAssignment assignment) {
  if (assignment.getContainersToKill() != null) {
    StringBuilder sb = new StringBuilder("Killing containers: [");

    for (RMContainer c : assignment.getContainersToKill()) {
      FiCaSchedulerApp application = csContext.getApplicationAttempt(
          c.getApplicationAttemptId());
      LeafQueue q = application.getCSLeafQueue();
      q.completedContainer(clusterResource, application, node, c, SchedulerUtils
              .createPreemptedContainerStatus(c.getContainerId(),
                  SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL,
          null, false);
      sb.append("(container=" + c.getContainerId() + " resource=" + c
          .getAllocatedResource() + ")");
    }

    sb.append("] for container=" + assignment.getAssignmentInformation()
        .getFirstAllocatedOrReservedContainerId() + " resource=" + assignment
        .getResource());
    LOG.info(sb.toString());

  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的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.RMContainerEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public synchronized void containerLaunchedOnNode(ContainerId containerId,
    NodeId nodeId) {
  // Inform the container
  RMContainer rmContainer = getRMContainer(containerId);
  if (rmContainer == null) {
    // Some unknown container sneaked into the system. Kill it.
    rmContext.getDispatcher().getEventHandler()
      .handle(new RMNodeCleanContainerEvent(nodeId, containerId));
    return;
  }

  rmContainer.handle(new RMContainerEvent(containerId,
      RMContainerEventType.LAUNCHED));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
public synchronized ContainersAndNMTokensAllocation
    pullNewlyAllocatedContainersAndNMTokens() {
  List<Container> returnContainerList =
      new ArrayList<Container>(newlyAllocatedContainers.size());
  List<NMToken> nmTokens = new ArrayList<NMToken>();
  for (Iterator<RMContainer> i = newlyAllocatedContainers.iterator(); i
    .hasNext();) {
    RMContainer rmContainer = i.next();
    Container container = rmContainer.getContainer();
    try {
      // create container token and NMToken altogether.
      container.setContainerToken(rmContext.getContainerTokenSecretManager()
        .createContainerToken(container.getId(), container.getNodeId(),
          getUser(), container.getResource(), container.getPriority(),
          rmContainer.getCreationTime(), this.logAggregationContext));
      NMToken nmToken =
          rmContext.getNMTokenSecretManager().createAndGetNMToken(getUser(),
            getApplicationAttemptId(), container);
      if (nmToken != null) {
        nmTokens.add(nmToken);
      }
    } catch (IllegalArgumentException e) {
      // DNS might be down, skip returning this container.
      LOG.error("Error trying to assign container token and NM token to" +
          " an allocated container " + container.getId(), e);
      continue;
    }
    returnContainerList.add(container);
    i.remove();
    rmContainer.handle(new RMContainerEvent(rmContainer.getContainerId(),
      RMContainerEventType.ACQUIRED));
  }
  return new ContainersAndNMTokensAllocation(returnContainerList, nmTokens);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
protected void warnOrKillContainer(RMContainer container) {
  ApplicationAttemptId appAttemptId = container.getApplicationAttemptId();
  FSAppAttempt app = getSchedulerApp(appAttemptId);
  FSLeafQueue queue = app.getQueue();
  LOG.info("Preempting container (prio=" + container.getContainer().getPriority() +
      "res=" + container.getContainer().getResource() +
      ") from queue " + queue.getName());
  
  Long time = app.getContainerPreemptionTime(container);

  if (time != null) {
    // if we asked for preemption more than maxWaitTimeBeforeKill ms ago,
    // proceed with kill
    if (time + waitTimeBeforeKill < getClock().getTime()) {
      ContainerStatus status =
        SchedulerUtils.createPreemptedContainerStatus(
          container.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER);

      recoverResourceRequestForContainer(container);
      // TODO: Not sure if this ever actually adds this to the list of cleanup
      // containers on the RMNode (see SchedulerNode.releaseContainer()).
      completedContainer(container, status, RMContainerEventType.KILL);
      LOG.info("Killing container" + container +
          " (after waiting for premption for " +
          (getClock().getTime() - time) + "ms)");
    }
  } else {
    // track the request in the FSAppAttempt itself
    app.addPreemption(container, getClock().getTime());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的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.RMContainerEventType; //导入依赖的package包/类
private synchronized void removeNode(RMNode rmNode) {
  FSSchedulerNode node = getFSSchedulerNode(rmNode.getNodeID());
  // This can occur when an UNHEALTHY node reconnects
  if (node == null) {
    return;
  }
  Resources.subtractFrom(clusterResource, rmNode.getTotalCapability());
  updateRootQueueMetrics();

  // Remove running containers
  List<RMContainer> runningContainers = node.getRunningContainers();
  for (RMContainer container : runningContainers) {
    completedContainer(container,
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(),
            SchedulerUtils.LOST_CONTAINER),
        RMContainerEventType.KILL);
  }

  // Remove reservations, if any
  RMContainer reservedContainer = node.getReservedContainer();
  if (reservedContainer != null) {
    completedContainer(reservedContainer,
        SchedulerUtils.createAbnormalContainerStatus(
            reservedContainer.getContainerId(),
            SchedulerUtils.LOST_CONTAINER),
        RMContainerEventType.KILL);
  }

  nodes.remove(rmNode.getNodeID());
  queueMgr.getRootQueue().setSteadyFairShare(clusterResource);
  queueMgr.getRootQueue().recomputeSteadyShares();
  updateMaximumAllocation(node, false);
  LOG.info("Removed node " + rmNode.getNodeAddress() +
      " cluster capacity: " + clusterResource);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
/**
 * Process a heartbeat update from a node.
 */
private synchronized void nodeUpdate(RMNode nm) {
  long start = getClock().getTime();
  if (LOG.isDebugEnabled()) {
    LOG.debug("nodeUpdate: " + nm + " cluster capacity: " + clusterResource);
  }
  eventLog.log("HEARTBEAT", nm.getHostName());
  FSSchedulerNode node = getFSSchedulerNode(nm.getNodeID());
  
  List<UpdatedContainerInfo> containerInfoList = nm.pullContainerUpdates();
  List<ContainerStatus> newlyLaunchedContainers = new ArrayList<ContainerStatus>();
  List<ContainerStatus> completedContainers = new ArrayList<ContainerStatus>();
  for(UpdatedContainerInfo containerInfo : containerInfoList) {
    newlyLaunchedContainers.addAll(containerInfo.getNewlyLaunchedContainers());
    completedContainers.addAll(containerInfo.getCompletedContainers());
  } 
  // Processing the newly launched containers
  for (ContainerStatus launchedContainer : newlyLaunchedContainers) {
    containerLaunchedOnNode(launchedContainer.getContainerId(), node);
  }

  // Process completed containers
  for (ContainerStatus completedContainer : completedContainers) {
    ContainerId containerId = completedContainer.getContainerId();
    LOG.debug("Container FINISHED: " + containerId);
    completedContainer(getRMContainer(containerId),
        completedContainer, RMContainerEventType.FINISHED);
  }

  if (continuousSchedulingEnabled) {
    if (!completedContainers.isEmpty()) {
      attemptScheduling(node);
    }
  } else {
    attemptScheduling(node);
  }

  long duration = getClock().getTime() - start;
  fsOpDurations.addNodeUpdateDuration(duration);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; //导入依赖的package包/类
synchronized public void containerCompleted(RMContainer rmContainer,
    ContainerStatus containerStatus, RMContainerEventType event) {
  
  Container container = rmContainer.getContainer();
  ContainerId containerId = container.getId();
  
  // Remove from the list of newly allocated containers if found
  newlyAllocatedContainers.remove(rmContainer);
  
  // Inform the container
  rmContainer.handle(
      new RMContainerFinishedEvent(
          containerId,
          containerStatus, 
          event)
      );
  LOG.info("Completed container: " + rmContainer.getContainerId() + 
      " in state: " + rmContainer.getState() + " event:" + event);
  
  // Remove from the list of containers
  liveContainers.remove(rmContainer.getContainerId());

  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.RELEASE_CONTAINER, "SchedulerApp", 
      getApplicationId(), containerId);
  
  // Update usage metrics 
  Resource containerResource = rmContainer.getContainer().getResource();
  queue.getMetrics().releaseResources(getUser(), 1, containerResource);
  this.attemptResourceUsage.decUsed(containerResource);

  // remove from preemption map if it is completed
  preemptionMap.remove(rmContainer);

  // Clear resource utilization metrics cache.
  lastMemoryAggregateAllocationUpdateTime = -1;
}