import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
@Test
public void testAddAndRemoveAppFromCapacityScheduler() throws Exception {
  CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
  setupQueueConfiguration(conf);
  conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
    ResourceScheduler.class);
  MockRM rm = new MockRM(conf);
  @SuppressWarnings("unchecked")
  AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode> cs =
      (AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode>) rm
        .getResourceScheduler();

  SchedulerApplication<SchedulerApplicationAttempt> app =
      TestSchedulerUtils.verifyAppAddedAndRemovedFromScheduler(
        cs.getSchedulerApplications(), cs, "a1");
  Assert.assertEquals("a1", app.getQueue().getQueueName());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
/**
 * Format is:
 * host1=partition;
 * host2=partition;
 */
private void mockSchedulerNodes(String schedulerNodesConfigStr)
    throws IOException {
  String[] nodesConfigStrArray = schedulerNodesConfigStr.split(";");
  for (String p : nodesConfigStrArray) {
    NodeId nodeId = NodeId.newInstance(p.substring(0, p.indexOf("=")), 1);
    String partition = p.substring(p.indexOf("=") + 1, p.length());

    SchedulerNode sn = mock(SchedulerNode.class);
    when(sn.getNodeID()).thenReturn(nodeId);
    when(sn.getPartition()).thenReturn(partition);
    nodeIdToSchedulerNodes.put(nodeId, sn);

    LOG.debug("add scheduler node, id=" + nodeId + ", partition=" + partition);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private void rollbackContainerResource(
    ContainerId containerId) {
  RMContainer rmContainer = getRMContainer(containerId);
  if (rmContainer == null) {
    LOG.info("Cannot rollback resource for container " + containerId +
        ". The container does not exist.");
    return;
  }
  FiCaSchedulerApp application = getCurrentAttemptForContainer(containerId);
  if (application == null) {
    LOG.info("Cannot rollback resource for container " + containerId +
        ". The application that the container belongs to does not exist.");
    return;
  }
  LOG.info("Roll back resource for container " + containerId);
  LeafQueue leafQueue = (LeafQueue) application.getQueue();
  synchronized(leafQueue) {
    SchedulerNode schedulerNode =
        getSchedulerNode(rmContainer.getAllocatedNode());
    SchedContainerChangeRequest decreaseRequest =
        new SchedContainerChangeRequest(this.rmContext, schedulerNode,
            rmContainer, rmContainer.getLastConfirmedResource());
    decreaseContainer(decreaseRequest, application);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
@Test
public void testAddAndRemoveAppFromFiFoScheduler() throws Exception {
  Configuration conf = new Configuration();
  conf.setClass(YarnConfiguration.RM_SCHEDULER, FifoScheduler.class,
    ResourceScheduler.class);
  MockRM rm = new MockRM(conf);
  @SuppressWarnings("unchecked")
  AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode> fs =
      (AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode>) rm
        .getResourceScheduler();
  TestSchedulerUtils.verifyAppAddedAndRemovedFromScheduler(
    fs.getSchedulerApplications(), fs, "queue");
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
@Test
public void testAddAndRemoveAppFromFairScheduler() throws Exception {
  AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode> scheduler =
      (AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode>) resourceManager
        .getResourceScheduler();
  TestSchedulerUtils.verifyAppAddedAndRemovedFromScheduler(
    scheduler.getSchedulerApplications(), scheduler, "default");
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void serviceInit(Configuration conf) throws Exception {
  ((AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode>)
      scheduler).init(conf);
  super.serviceInit(conf);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void serviceStart() throws Exception {
  ((AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode>)
      scheduler).start();
  super.serviceStart();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void serviceStop() throws Exception {
  ((AbstractYarnScheduler<SchedulerApplicationAttempt, SchedulerNode>)
      scheduler).stop();
  super.serviceStop();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private void addAMNodeToBlackList(NodeId nodeId) {
  SchedulerNode schedulerNode = scheduler.getSchedulerNode(nodeId);
  if (schedulerNode != null) {
    blacklistedNodesForAM.addNode(schedulerNode.getNodeName());
  } else {
    LOG.info(nodeId + " is not added to AM blacklist for "
        + applicationAttemptId + ", because it has been removed");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private synchronized void setReservation(SchedulerNode node) {
  String rackName = node.getRackName() == null ? "NULL" : node.getRackName();
  Set<String> rackReservations = reservations.get(rackName);
  if (rackReservations == null) {
    rackReservations = new HashSet<>();
    reservations.put(rackName, rackReservations);
  }
  rackReservations.add(node.getNodeName());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private synchronized void clearReservation(SchedulerNode node) {
  String rackName = node.getRackName() == null ? "NULL" : node.getRackName();
  Set<String> rackReservations = reservations.get(rackName);
  if (rackReservations != null) {
    rackReservations.remove(node.getNodeName());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private CSAssignment allocateIncreaseRequestFromReservedContainer(
    SchedulerNode node, Resource cluster,
    SchedContainerChangeRequest increaseRequest) {
  if (Resources.fitsIn(rc, cluster, increaseRequest.getDeltaCapacity(),
      node.getAvailableResource())) {
    // OK, we can allocate this increase request
    // Unreserve it first
    application.unreserve(increaseRequest.getPriority(),
        (FiCaSchedulerNode) node, increaseRequest.getRMContainer());
    
    // Notify application
    application.increaseContainer(increaseRequest);
    
    // Notify node
    node.increaseContainer(increaseRequest.getContainerId(),
        increaseRequest.getDeltaCapacity());

    return createSuccessfullyIncreasedCSAssignment(increaseRequest, true);
  } else {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Failed to allocate reserved increase request:"
          + increaseRequest.toString()
          + ". There's no enough available resource");
    }
    
    // We still cannot allocate this container, will wait for next turn
    return CSAssignment.SKIP_ASSIGNMENT;
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private void verifyAvailableResourceOfSchedulerNode(MockRM rm, NodeId nodeId,
    int expectedMemory) {
  CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
  SchedulerNode node = cs.getNode(nodeId);
  Assert
      .assertEquals(expectedMemory, node.getAvailableResource().getMemory());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private void setResourceAndNodeDetails() {
  when(mCS.getClusterResource()).thenReturn(clusterResources);
  when(lm.getResourceByLabel(anyString(), any(Resource.class))).thenReturn(
      clusterResources);

  SchedulerNode mNode = mock(SchedulerNode.class);
  when(mNode.getPartition()).thenReturn(RMNodeLabelsManager.NO_LABEL);
  when(mCS.getSchedulerNode(any(NodeId.class))).thenReturn(mNode);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
private void subtractResourcesOnBlacklistedNodes(
    Resource availableResources) {
  if (appSchedulingInfo.getAndResetBlacklistChanged()) {
    blacklistNodeIds.clear();
    scheduler.addBlacklistedNodeIdsToList(this, blacklistNodeIds);
  }
  for (NodeId nodeId: blacklistNodeIds) {
    SchedulerNode node = scheduler.getSchedulerNode(nodeId);
    if (node != null) {
      Resources.subtractFromNonNegative(availableResources,
          node.getAvailableResource());
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; //导入依赖的package包/类
public static void deductPreemptableResourcesBasedSelectedCandidates(
    CapacitySchedulerPreemptionContext context,
    Map<ApplicationAttemptId, Set<RMContainer>> selectedCandidates) {
  for (Set<RMContainer> containers : selectedCandidates.values()) {
    for (RMContainer c : containers) {
      SchedulerNode schedulerNode = context.getScheduler()
          .getSchedulerNode(c.getAllocatedNode());
      if (null == schedulerNode) {
        continue;
      }

      String partition = schedulerNode.getPartition();
      String queue = c.getQueueName();
      TempQueuePerPartition tq = context.getQueueByPartition(queue,
          partition);

      Resource res = c.getReservedResource();
      if (null == res) {
        res = c.getAllocatedResource();
      }

      if (null != res) {
        tq.deductActuallyToBePreempted(context.getResourceCalculator(),
            tq.totalPartitionResource, res);
        Collection<TempAppPerPartition> tas = tq.getApps();
        if (null == tas || tas.isEmpty()) {
          continue;
        }

        deductPreemptableResourcePerApp(context, tq.totalPartitionResource,
            tas, res, partition);
      }
    }
  }
}