import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
/**
 * Create an {@link AMRMTokenSecretManager}
 */
public AMRMTokenSecretManager(Configuration conf, RMContext rmContext) {
  this.rmContext = rmContext;
  this.timer = new Timer();
  this.rollingInterval =
      conf
        .getLong(
          YarnConfiguration.RM_AMRM_TOKEN_MASTER_KEY_ROLLING_INTERVAL_SECS,
          YarnConfiguration.DEFAULT_RM_AMRM_TOKEN_MASTER_KEY_ROLLING_INTERVAL_SECS) * 1000;
  // Adding delay = 1.5 * expiry interval makes sure that all active AMs get
  // the updated shared-key.
  this.activationDelay =
      (long) (conf.getLong(YarnConfiguration.RM_AM_EXPIRY_INTERVAL_MS,
          YarnConfiguration.DEFAULT_RM_AM_EXPIRY_INTERVAL_MS) * 1.5);
  LOG.info("AMRMTokenKeyRollingInterval: " + this.rollingInterval
      + "ms and AMRMTokenKeyActivationDelay: " + this.activationDelay + " ms");
  if (rollingInterval <= activationDelay * 2) {
    throw new IllegalArgumentException(
        YarnConfiguration.RM_AMRM_TOKEN_MASTER_KEY_ROLLING_INTERVAL_SECS
            + " should be more than 3 X "
            + YarnConfiguration.RM_AM_EXPIRY_INTERVAL_MS);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public RMContainerImpl(Container container,
    ApplicationAttemptId appAttemptId, NodeId nodeId,
    String user, RMContext rmContext, long creationTime) {
  this.stateMachine = stateMachineFactory.make(this);
  this.containerId = container.getId();
  this.nodeId = nodeId;
  this.container = container;
  this.appAttemptId = appAttemptId;
  this.user = user;
  this.creationTime = creationTime;
  this.rmContext = rmContext;
  this.eventHandler = rmContext.getDispatcher().getEventHandler();
  this.containerAllocationExpirer = rmContext.getContainerAllocationExpirer();
  this.isAMContainer = false;
  this.resourceRequests = null;

  ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  this.readLock = lock.readLock();
  this.writeLock = lock.writeLock();

  rmContext.getRMApplicationHistoryWriter().containerStarted(this);
  rmContext.getSystemMetricsPublisher().containerCreated(
      this, this.creationTime);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public RMNodeImpl(NodeId nodeId, RMContext context, String hostName,
    int cmPort, int httpPort, Node node, Resource capability, String nodeManagerVersion) {
  this.nodeId = nodeId;
  this.context = context;
  this.hostName = hostName;
  this.commandPort = cmPort;
  this.httpPort = httpPort;
  this.totalCapability = capability; 
  this.nodeAddress = hostName + ":" + cmPort;
  this.httpAddress = hostName + ":" + httpPort;
  this.node = node;
  this.healthReport = "Healthy";
  this.lastHealthReportTime = System.currentTimeMillis();
  this.nodeManagerVersion = nodeManagerVersion;

  this.latestNodeHeartBeatResponse.setResponseId(0);

  ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  this.readLock = lock.readLock();
  this.writeLock = lock.writeLock();

  this.stateMachine = stateMachineFactory.make(this);
  
  this.nodeUpdateQueue = new ConcurrentLinkedQueue<UpdatedContainerInfo>();  
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
private static void handleRunningAppOnNode(RMNodeImpl rmNode,
    RMContext context, ApplicationId appId, NodeId nodeId) {
  RMApp app = context.getRMApps().get(appId);

  // if we failed getting app by appId, maybe something wrong happened, just
  // add the app to the finishedApplications list so that the app can be
  // cleaned up on the NM
  if (null == app) {
    LOG.warn("Cannot get RMApp by appId=" + appId
        + ", just added it to finishedApplications list for cleanup");
    rmNode.finishedApplications.add(appId);
    return;
  }

  context.getDispatcher().getEventHandler()
      .handle(new RMAppRunningOnNodeEvent(appId, nodeId));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public RMAppAttemptImpl(ApplicationAttemptId appAttemptId,
    RMContext rmContext, YarnScheduler scheduler,
    ApplicationMasterService masterService,
    ApplicationSubmissionContext submissionContext,
    Configuration conf, boolean maybeLastAttempt, ResourceRequest amReq) {
  this.conf = conf;
  this.applicationAttemptId = appAttemptId;
  this.rmContext = rmContext;
  this.eventHandler = rmContext.getDispatcher().getEventHandler();
  this.submissionContext = submissionContext;
  this.scheduler = scheduler;
  this.masterService = masterService;

  ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  this.readLock = lock.readLock();
  this.writeLock = lock.writeLock();

  this.proxiedTrackingUrl = generateProxyUriWithScheme();
  this.maybeLastAttempt = maybeLastAttempt;
  this.stateMachine = stateMachineFactory.make(this);

  this.attemptMetrics =
      new RMAppAttemptMetrics(applicationAttemptId, rmContext);
  
  this.amReq = amReq;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public SchedulerApplicationAttempt(ApplicationAttemptId applicationAttemptId, 
    String user, Queue queue, ActiveUsersManager activeUsersManager,
    RMContext rmContext) {
  Preconditions.checkNotNull(rmContext, "RMContext should not be null");
  this.rmContext = rmContext;
  this.appSchedulingInfo = 
      new AppSchedulingInfo(applicationAttemptId, user, queue,  
          activeUsersManager, rmContext.getEpoch());
  this.queue = queue;
  this.pendingRelease = new HashSet<ContainerId>();
  this.attemptId = applicationAttemptId;
  if (rmContext.getRMApps() != null &&
      rmContext.getRMApps()
          .containsKey(applicationAttemptId.getApplicationId())) {
    ApplicationSubmissionContext appSubmissionContext =
        rmContext.getRMApps().get(applicationAttemptId.getApplicationId())
            .getApplicationSubmissionContext();
    if (appSubmissionContext != null) {
      unmanagedAM = appSubmissionContext.getUnmanagedAM();
      this.logAggregationContext =
          appSubmissionContext.getLogAggregationContext();
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public static void normalizeAndValidateRequest(ResourceRequest resReq,
    Resource maximumResource, String queueName, YarnScheduler scheduler,
    boolean isRecovery, RMContext rmContext, QueueInfo queueInfo)
    throws InvalidResourceRequestException {

  if (queueInfo == null) {
    try {
      queueInfo = scheduler.getQueueInfo(queueName, false, false);
    } catch (IOException e) {
      // it is possible queue cannot get when queue mapping is set, just ignore
      // the queueInfo here, and move forward
    }
  }
  SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo);
  if (!isRecovery) {
    validateResourceRequest(resReq, maximumResource, queueInfo, rmContext);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Override
public synchronized void
reinitialize(Configuration conf, RMContext rmContext) throws IOException {
  Configuration configuration = new Configuration(conf);
  CapacitySchedulerConfiguration oldConf = this.conf;
  this.conf = loadCapacitySchedulerConfiguration(configuration);
  validateConf(this.conf);
  try {
    LOG.info("Re-initializing queues...");
    refreshMaximumAllocation(this.conf.getMaximumAllocation());
    reinitializeQueues(this.conf);
  } catch (Throwable t) {
    this.conf = oldConf;
    refreshMaximumAllocation(this.conf.getMaximumAllocation());
    throw new IOException("Failed to re-init queues", t);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  counter = new AtomicInteger(0);
  conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION,
      "kerberos");
  UserGroupInformation.setConfiguration(conf);
  eventQueue = new LinkedBlockingQueue<Event>();
  dispatcher = new AsyncDispatcher(eventQueue);
  Renewer.reset();
  delegationTokenRenewer = createNewDelegationTokenRenewer(conf, counter);
  RMContext mockContext =  mock(RMContext.class);
  ClientRMService mockClientRMService = mock(ClientRMService.class);
  when(mockContext.getSystemCredentialsForApps()).thenReturn(
    new ConcurrentHashMap<ApplicationId, ByteBuffer>());
  when(mockContext.getDelegationTokenRenewer()).thenReturn(
      delegationTokenRenewer);
  when(mockContext.getDispatcher()).thenReturn(dispatcher);
  when(mockContext.getClientRMService()).thenReturn(mockClientRMService);
  InetSocketAddress sockAddr =
      InetSocketAddress.createUnresolved("localhost", 1234);
  when(mockClientRMService.getBindAddress()).thenReturn(sockAddr);
  delegationTokenRenewer.setRMContext(mockContext);
  delegationTokenRenewer.init(conf);
  delegationTokenRenewer.start();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public static CapacityScheduler mockCapacityScheduler() throws IOException {
  // stolen from TestCapacityScheduler
  CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
  setupQueueConfiguration(conf);

  CapacityScheduler cs = new CapacityScheduler();
  cs.setConf(new YarnConfiguration());
  RMContext rmContext = new RMContextImpl(null, null, null, null, null,
      null, new RMContainerTokenSecretManager(conf),
      new NMTokenSecretManagerInRM(conf),
      new ClientToAMTokenSecretManagerInRM(), null);
  rmContext.setNodeLabelManager(new NullRMNodeLabelsManager());
  cs.setRMContext(rmContext);
  cs.init(conf);
  return cs;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  final RMContext mockRMContext =
      TestRMWebApp.mockRMContext(3, numberOfRacks, numberOfNodesPerRack,
        8 * TestRMWebApp.GiB);
  injector =
      WebAppTests.createMockInjector(RMContext.class, mockRMContext,
        new Module() {
          @Override
          public void configure(Binder binder) {
            try {
              binder.bind(ResourceManager.class).toInstance(
                TestRMWebApp.mockRm(mockRMContext));
            } catch (IOException e) {
              throw new IllegalStateException(e);
            }
          }
        });
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public static FairScheduler setupFairScheduler(
    ReservationSystemTestUtil testUtil,
    RMContext rmContext, Configuration conf, int numContainers) throws
    IOException {
  FairScheduler scheduler = new FairScheduler();
  scheduler.setRMContext(rmContext);

  when(rmContext.getScheduler()).thenReturn(scheduler);

  scheduler.init(conf);
  scheduler.start();
  scheduler.reinitialize(conf, rmContext);


  Resource resource = testUtil.calculateClusterResource(numContainers);
  RMNode node1 = MockNodes.newNodeInfo(1, resource, 1, "127.0.0.1");
  NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
  scheduler.handle(nodeEvent1);
  return scheduler;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public CapacityScheduler mockCapacityScheduler(int numContainers)
    throws IOException {
  // stolen from TestCapacityScheduler
  CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
  setupQueueConfiguration(conf);

  CapacityScheduler cs = Mockito.spy(new CapacityScheduler());
  cs.setConf(new YarnConfiguration());

  RMContext mockRmContext = createRMContext(conf);

  cs.setRMContext(mockRmContext);
  try {
    cs.serviceInit(conf);
  } catch (Exception e) {
    Assert.fail(e.getMessage());
  }

  initializeRMContext(numContainers, cs, mockRmContext);
  return cs;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Test
public void testFairReservationSystemInitialize() throws IOException {
  ReservationSystemTestUtil.setupFSAllocationFile(ALLOC_FILE);

  ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
  
  // Setup
  RMContext mockRMContext = testUtil.createRMContext(conf);
  scheduler = ReservationSystemTestUtil.setupFairScheduler(testUtil,
      mockRMContext, conf, 10);

  FairReservationSystem reservationSystem = new FairReservationSystem();
  reservationSystem.setRMContext(mockRMContext);

  try {
    reservationSystem.reinitialize(scheduler.getConf(), mockRMContext);
  } catch (YarnException e) {
    Assert.fail(e.getMessage());
  }

  ReservationSystemTestUtil.validateReservationQueue(reservationSystem,
      testUtil.getFullReservationQueueName());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  Configuration conf = new Configuration();
  clock = new TestFairScheduler.MockClock();
  scheduler = mock(FairScheduler.class);
  when(scheduler.getConf()).thenReturn(
      new FairSchedulerConfiguration(conf));
  when(scheduler.getClock()).thenReturn(clock);
  AllocationConfiguration allocConf = new AllocationConfiguration(
      conf);
  when(scheduler.getAllocationConfiguration()).thenReturn(allocConf);
  
  queueManager = new QueueManager(scheduler);
  queueManager.initialize(conf);
  queueMaxApps = allocConf.queueMaxApps;
  userMaxApps = allocConf.userMaxApps;
  maxAppsEnforcer = new MaxRunningAppsEnforcer(scheduler);
  appNum = 0;
  rmContext = mock(RMContext.class);
  when(rmContext.getEpoch()).thenReturn(0L);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Test
/**
 * Ensure that when negative paramaters are given (signaling delay scheduling
 * no tin use), the least restrictive locality level is returned.
 */
public void testLocalityLevelWithoutDelays() {
  FSLeafQueue queue = Mockito.mock(FSLeafQueue.class);
  Priority prio = Mockito.mock(Priority.class);
  Mockito.when(prio.getPriority()).thenReturn(1);

  RMContext rmContext = resourceManager.getRMContext();
  ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1);
  FSAppAttempt schedulerApp =
      new FSAppAttempt(scheduler, applicationAttemptId, "user1", queue ,
          null, rmContext);
  assertEquals(NodeType.OFF_SWITCH, schedulerApp.getAllowedLocalityLevel(
      prio, 10, -1.0, -1.0));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Before
public void setup() throws IOException {
  // setup a context / conf
  csConf = new CapacitySchedulerConfiguration();
  YarnConfiguration conf = new YarnConfiguration();
  csContext = mock(CapacitySchedulerContext.class);
  when(csContext.getConfiguration()).thenReturn(csConf);
  when(csContext.getConf()).thenReturn(conf);
  when(csContext.getMinimumResourceCapability()).thenReturn(
      Resources.createResource(GB, 1, 1));
  when(csContext.getMaximumResourceCapability()).thenReturn(
      Resources.createResource(16 * GB, 32, 32));
  when(csContext.getClusterResource()).thenReturn(
      Resources.createResource(100 * 16 * GB, 100 * 32, 100 * 32));
  when(csContext.getResourceCalculator()).thenReturn(resourceCalculator);
  
  RMContext mockRMContext = TestUtils.getMockRMContext();
  when(csContext.getRMContext()).thenReturn(mockRMContext);

  // create a queue
  PlanQueue pq = new PlanQueue(csContext, "root", null, null);
  reservationQueue = new ReservationQueue(csContext, "a", pq);

}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  resourceManager = new ResourceManager() {
    @Override
    protected RMNodeLabelsManager createNodeLabelManager() {
      RMNodeLabelsManager mgr = new NullRMNodeLabelsManager();
      mgr.init(getConfig());
      return mgr;
    }
  };
  CapacitySchedulerConfiguration csConf 
     = new CapacitySchedulerConfiguration();
  setupQueueConfiguration(csConf);
  YarnConfiguration conf = new YarnConfiguration(csConf);
  conf.setClass(YarnConfiguration.RM_SCHEDULER, 
      CapacityScheduler.class, ResourceScheduler.class);
  resourceManager.init(conf);
  resourceManager.getRMContext().getContainerTokenSecretManager().rollMasterKey();
  resourceManager.getRMContext().getNMTokenSecretManager().rollMasterKey();
  ((AsyncDispatcher)resourceManager.getRMContext().getDispatcher()).start();
  mockContext = mock(RMContext.class);
  when(mockContext.getConfigurationProvider()).thenReturn(
      new LocalConfigurationProvider());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public FifoSchedulerInfo(final ResourceManager rm) {

    RMContext rmContext = rm.getRMContext();

    FifoScheduler fs = (FifoScheduler) rm.getResourceScheduler();
    qName = fs.getQueueInfo("", false, false).getQueueName();
    QueueInfo qInfo = fs.getQueueInfo(qName, true, true);

    this.usedCapacity = qInfo.getCurrentCapacity();
    this.capacity = qInfo.getCapacity();
    this.minQueueMemoryCapacity = fs.getMinimumResourceCapability().getMemory();
    this.maxQueueMemoryCapacity = fs.getMaximumResourceCapability().getMemory();
    this.qstate = qInfo.getQueueState();

    this.numNodes = rmContext.getRMNodes().size();
    this.usedNodeCapacity = 0;
    this.availNodeCapacity = 0;
    this.totalNodeCapacity = 0;
    this.numContainers = 0;

    for (RMNode ni : rmContext.getRMNodes().values()) {
      SchedulerNodeReport report = fs.getNodeReport(ni.getNodeID());
      this.usedNodeCapacity += report.getUsedResource().getMemory();
      this.availNodeCapacity += report.getAvailableResource().getMemory();
      this.totalNodeCapacity += ni.getTotalCapability().getMemory();
      this.numContainers += fs.getNodeReport(ni.getNodeID()).getNumContainers();
    }
  }
 

import org.apache.hadoop.yarn.server.resourcemanager.RMContext; //导入依赖的package包/类
public AMLauncher(RMContext rmContext, RMAppAttempt application,
    AMLauncherEventType eventType, Configuration conf) {
  this.application = application;
  this.conf = conf;
  this.eventType = eventType;
  this.rmContext = rmContext;
  this.handler = rmContext.getDispatcher().getEventHandler();
  this.masterContainer = application.getMasterContainer();
}