import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Inject
public FairSchedulerAppsBlock(ResourceManager rm, ViewContext ctx,
    Configuration conf) {
  super(ctx);
  FairScheduler scheduler = (FairScheduler) rm.getResourceScheduler();
  fsinfo = new FairSchedulerInfo(scheduler);
  apps = new ConcurrentHashMap<ApplicationId, RMApp>();
  for (Map.Entry<ApplicationId, RMApp> entry : rm.getRMContext().getRMApps()
      .entrySet()) {
    if (!(RMAppState.NEW.equals(entry.getValue().getState())
        || RMAppState.NEW_SAVING.equals(entry.getValue().getState())
        || RMAppState.SUBMITTED.equals(entry.getValue().getState()))) {
      apps.put(entry.getKey(), entry.getValue());
    }
  }
  this.conf = conf;
  this.rm = rm;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
private ConcurrentHashMap<ApplicationId, RMApp> getRMApps(
    RMContext rmContext, YarnScheduler yarnScheduler) {
  ConcurrentHashMap<ApplicationId, RMApp> apps = 
    new ConcurrentHashMap<ApplicationId, RMApp>();
  ApplicationId applicationId1 = getApplicationId(1);
  ApplicationId applicationId2 = getApplicationId(2);
  ApplicationId applicationId3 = getApplicationId(3);
  YarnConfiguration config = new YarnConfiguration();
  apps.put(applicationId1, getRMApp(rmContext, yarnScheduler, applicationId1,
      config, "testqueue", 10, 3, 3));
  apps.put(applicationId2, getRMApp(rmContext, yarnScheduler, applicationId2,
      config, "a", 20, 2, 2));
  apps.put(applicationId3, getRMApp(rmContext, yarnScheduler, applicationId3,
      config, "testqueue", 40, 5, 5));
  return apps;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void
    testHandleRMHABeforeSubmitApplicationCallWithSavedApplicationState()
        throws Exception {
  // start two RMs, and transit rm1 to active, rm2 to standby
  startRMs();

  // get a new applicationId from rm1
  ApplicationId appId = rm1.getNewAppId().getApplicationId();

  // Do the failover
  explicitFailover();

  // submit the application with previous assigned applicationId
  // to current active rm: rm2
  RMApp app1 =
      rm2.submitApp(200, "", UserGroupInformation
          .getCurrentUser().getShortUserName(), null, false, null,
          configuration.getInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS,
              YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS), null, null,
          false, false, true, appId);

  // verify application submission
  verifySubmitApp(rm2, app1, appId);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testAssignToQueue() throws Exception {
  conf.set(FairSchedulerConfiguration.USER_AS_DEFAULT_QUEUE, "true");
  scheduler.init(conf);
  scheduler.start();
  scheduler.reinitialize(conf, resourceManager.getRMContext());

  RMApp rmApp1 = new MockRMApp(0, 0, RMAppState.NEW);
  RMApp rmApp2 = new MockRMApp(1, 1, RMAppState.NEW);
  
  FSLeafQueue queue1 = scheduler.assignToQueue(rmApp1, "default", "asterix");
  FSLeafQueue queue2 = scheduler.assignToQueue(rmApp2, "notdefault", "obelix");
  
  // assert FSLeafQueue's name is the correct name is the one set in the RMApp
  assertEquals(rmApp1.getQueue(), queue1.getName());
  assertEquals("root.asterix", rmApp1.getQueue());
  assertEquals(rmApp2.getQueue(), queue2.getName());
  assertEquals("root.notdefault", rmApp2.getQueue());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public void appAttemptFinished(RMAppAttempt appAttempt,
    RMAppAttemptState appAttemtpState, RMApp app, long finishedTime) {
  if (publishSystemMetrics) {
    dispatcher.getEventHandler().handle(
        new AppAttemptFinishedEvent(
            appAttempt.getAppAttemptId(),
            appAttempt.getTrackingUrl(),
            appAttempt.getOriginalTrackingUrl(),
            appAttempt.getDiagnostics(),
            // app will get the final status from app attempt, or create one
            // based on app state if it doesn't exist
            app.getFinalApplicationStatus(),
            RMServerUtils.createApplicationAttemptState(appAttemtpState),
            finishedTime));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的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.rmapp.RMApp; //导入依赖的package包/类
@Override
public void handle(RMAppAttemptEvent event) {
  ApplicationAttemptId appAttemptID = event.getApplicationAttemptId();
  ApplicationId appAttemptId = appAttemptID.getApplicationId();
  RMApp rmApp = this.rmContext.getRMApps().get(appAttemptId);
  if (rmApp != null) {
    RMAppAttempt rmAppAttempt = rmApp.getRMAppAttempt(appAttemptID);
    if (rmAppAttempt != null) {
      try {
        rmAppAttempt.handle(event);
      } catch (Throwable t) {
        LOG.error("Error in handling event type " + event.getType()
            + " for applicationAttempt " + appAttemptId, t);
      }
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testMoveRejectedByScheduler() throws Exception {
  failMove = true;
  
  // Submit application
  Application application = new Application("user1", resourceManager);
  application.submit();

  // Wait for app to be accepted
  RMApp app = resourceManager.rmContext.getRMApps()
          .get(application.getApplicationId());
  while (app.getState() != RMAppState.ACCEPTED) {
    Thread.sleep(100);
  }

  ClientRMService clientRMService = resourceManager.getClientRMService();
  try {
    // FIFO scheduler does not support moves
    clientRMService.moveApplicationAcrossQueues(
        MoveApplicationAcrossQueuesRequest.newInstance(
            application.getApplicationId(), "newqueue"));
    fail("Should have hit exception");
  } catch (YarnException ex) {
    assertEquals("Move not supported", ex.getCause().getMessage());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testForceKillNonExistingApplication() throws YarnException {
  RMContext rmContext = mock(RMContext.class);
  when(rmContext.getRMApps()).thenReturn(
      new ConcurrentHashMap<ApplicationId, RMApp>());
  ClientRMService rmService = new ClientRMService(rmContext, null, null,
      null, null, null);
  ApplicationId applicationId =
      BuilderUtils.newApplicationId(System.currentTimeMillis(), 0);
  KillApplicationRequest request =
      KillApplicationRequest.newInstance(applicationId);
  try {
    rmService.forceKillApplication(request);
    Assert.fail();
  } catch (ApplicationNotFoundException ex) {
    Assert.assertEquals(ex.getMessage(),
        "Trying to kill an absent " +
            "application " + request.getApplicationId());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testStateStoreAppLimitLargerThanMemoryAppLimit() {
  long now = System.currentTimeMillis();
  RMContext rmContext = mockRMContext(10, now - 20000);
  Configuration conf = new YarnConfiguration();
  int maxAppsInMemory = 8;
  conf.setInt(YarnConfiguration.RM_MAX_COMPLETED_APPLICATIONS, maxAppsInMemory);
  // larger than maxCompletedAppsInMemory, reset to RM_MAX_COMPLETED_APPLICATIONS.
  conf.setInt(YarnConfiguration.RM_STATE_STORE_MAX_COMPLETED_APPLICATIONS, 1000);
  TestRMAppManager appMonitor = new TestRMAppManager(rmContext, conf);

  addToCompletedApps(appMonitor, rmContext);
  Assert.assertEquals("Number of completed apps incorrect", 10,
      appMonitor.getCompletedAppsListSize());
  appMonitor.checkAppNumCompletedLimit();

  int numRemoveApps = 10 - maxAppsInMemory;
  Assert.assertEquals("Number of apps incorrect after # completed check",
    maxAppsInMemory, rmContext.getRMApps().size());
  Assert.assertEquals("Number of completed apps incorrect after check",
    maxAppsInMemory, appMonitor.getCompletedAppsListSize());
  verify(rmContext.getStateStore(), times(numRemoveApps)).removeApplication(
    isA(RMApp.class));
  Assert.assertEquals(maxAppsInMemory,
    appMonitor.getCompletedAppsInStateStore());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testRMAppSubmit() throws Exception {
  appMonitor.submitApplication(asContext, "test");
  RMApp app = rmContext.getRMApps().get(appId);
  Assert.assertNotNull("app is null", app);
  Assert.assertEquals("app id doesn't match", appId, app.getApplicationId());
  Assert.assertEquals("app state doesn't match", RMAppState.NEW, app.getState());
  verify(metricsPublisher).appACLsUpdated(
      any(RMApp.class), any(String.class), anyLong());

  // wait for event to be processed
  int timeoutSecs = 0;
  while ((getAppEventType() == RMAppEventType.KILL) && 
      timeoutSecs++ < 20) {
    Thread.sleep(1000);
  }
  Assert.assertEquals("app event type sent is wrong", RMAppEventType.START,
      getAppEventType());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test (timeout = 30000)
public void testRMAppSubmitDuplicateApplicationId() throws Exception {
  ApplicationId appId = MockApps.newAppID(0);
  asContext.setApplicationId(appId);
  RMApp appOrig = rmContext.getRMApps().get(appId);
  Assert.assertTrue("app name matches but shouldn't", "testApp1" != appOrig.getName());

  // our testApp1 should be rejected and original app with same id should be left in place
  try {
    appMonitor.submitApplication(asContext, "test");
    Assert.fail("Exception is expected when applicationId is duplicate.");
  } catch (YarnException e) {
    Assert.assertTrue("The thrown exception is not the expectd one.",
        e.getMessage().contains("Cannot add a duplicate!"));
  }

  // make sure original app didn't get removed
  RMApp app = rmContext.getRMApps().get(appId);
  Assert.assertNotNull("app is null", app);
  Assert.assertEquals("app id doesn't match", appId, app.getApplicationId());
  Assert.assertEquals("app state doesn't match", RMAppState.FINISHED, app.getState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test (timeout = 20000)
public void testKillAppWhenFailoverHappensAtRunningState()
    throws Exception {
  startRMs();
  MockNM nm1 = new MockNM("127.0.0.1:1234", 15120,
      rm1.getResourceTrackerService());
  nm1.registerNode();

  // create app and launch the AM
  RMApp app0 = rm1.submitApp(200);
  MockAM am0 = launchAM(app0, rm1, nm1);

  // failover and kill application
  // The application is at RUNNING State when failOver happens.
  // Since RMStateStore has already saved ApplicationState, the active RM
  // will load the ApplicationState. After that, the application will be at
  // ACCEPTED State. Because the application is not at Final State,
  // KillApplicationResponse.getIsKillCompleted is expected to return false.
  failOverAndKillApp(app0.getApplicationId(),
      am0.getApplicationAttemptId(), RMAppState.RUNNING,
      RMAppAttemptState.RUNNING, RMAppState.ACCEPTED);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testAppsXML() throws JSONException, Exception {
  rm.start();
  MockNM amNodeManager = rm.registerNode("127.0.0.1:1234", 2048);
  RMApp app1 = rm.submitApp(CONTAINER_MB, "testwordcount", "user1");
  amNodeManager.nodeHeartbeat(true);
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("cluster")
      .path("apps").accept(MediaType.APPLICATION_XML)
      .get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_XML_TYPE, response.getType());
  String xml = response.getEntity(String.class);
  DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
  DocumentBuilder db = dbf.newDocumentBuilder();
  InputSource is = new InputSource();
  is.setCharacterStream(new StringReader(xml));
  Document dom = db.parse(is);
  NodeList nodesApps = dom.getElementsByTagName("apps");
  assertEquals("incorrect number of elements", 1, nodesApps.getLength());
  NodeList nodes = dom.getElementsByTagName("app");
  assertEquals("incorrect number of elements", 1, nodes.getLength());
  verifyAppsXML(nodes, app1);
  rm.stop();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testKillApplication() throws Exception {
  MockRM rm = new MockRM();
  rm.start();
  RMApp app = rm.submitApp(2000);

  Configuration conf = new Configuration();
  @SuppressWarnings("resource")
  final YarnClient client = new MockYarnClient();
  client.init(conf);
  client.start();

  client.killApplication(app.getApplicationId());
  verify(((MockYarnClient) client).getRMClient(), times(2))
    .forceKillApplication(any(KillApplicationRequest.class));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test (timeout = 60000)
public void testInvalidatedAMHostPortOnAMRestart() throws Exception {
  MockRM rm1 = new MockRM(conf);
  rm1.start();
  MockNM nm1 =
      new MockNM("127.0.0.1:1234", 15120, rm1.getResourceTrackerService());
  nm1.registerNode();

  // a failed app
  RMApp app2 = rm1.submitApp(200);
  MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);
  nm1
    .nodeHeartbeat(am2.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
  am2.waitForState(RMAppAttemptState.FAILED);
  rm1.waitForState(app2.getApplicationId(), RMAppState.ACCEPTED);

  // before new attempt is launched, the app report returns the invalid AM
  // host and port.
  GetApplicationReportRequest request1 =
      GetApplicationReportRequest.newInstance(app2.getApplicationId());
  ApplicationReport report1 =
      rm1.getClientRMService().getApplicationReport(request1)
        .getApplicationReport();
  Assert.assertEquals("N/A", report1.getHost());
  Assert.assertEquals(-1, report1.getRpcPort());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
public void testAppAttemptsHelper(String path, RMApp app, String media)
    throws JSONException, Exception {
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("cluster")
      .path("apps").path(path).path("appattempts").accept(media)
      .get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
  JSONObject json = response.getEntity(JSONObject.class);
  assertEquals("incorrect number of elements", 1, json.length());
  JSONObject jsonAppAttempts = json.getJSONObject("appAttempts");
  assertEquals("incorrect number of elements", 1, jsonAppAttempts.length());
  JSONArray jsonArray = jsonAppAttempts.getJSONArray("appAttempt");

  Collection<RMAppAttempt> attempts = app.getAppAttempts().values();
  assertEquals("incorrect number of elements", attempts.size(),
      jsonArray.length());

  // Verify these parallel arrays are the same
  int i = 0;
  for (RMAppAttempt attempt : attempts) {
    verifyAppAttemptsInfo(jsonArray.getJSONObject(i), attempt, app.getUser());
    ++i;
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test
public void testAssignToNonLeafQueueReturnsNull() throws Exception {
  conf.set(FairSchedulerConfiguration.USER_AS_DEFAULT_QUEUE, "true");
  scheduler.init(conf);
  scheduler.start();
  scheduler.reinitialize(conf, resourceManager.getRMContext());

  scheduler.getQueueManager().getLeafQueue("root.child1.granchild", true);
  scheduler.getQueueManager().getLeafQueue("root.child2", true);

  RMApp rmApp1 = new MockRMApp(0, 0, RMAppState.NEW);
  RMApp rmApp2 = new MockRMApp(1, 1, RMAppState.NEW);

  // Trying to assign to non leaf queue would return null
  assertNull(scheduler.assignToQueue(rmApp1, "root.child1", "tintin"));
  assertNotNull(scheduler.assignToQueue(rmApp2, "root.child2", "snowy"));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
@Test (timeout = 30000)
public void testAppWithNoContainers() throws Exception {
  Logger rootLogger = LogManager.getRootLogger();
  rootLogger.setLevel(Level.DEBUG);
  MockRM rm = new MockRM(conf);
  rm.start();
  MockNM nm1 = rm.registerNode("h1:1234", 5120);
  
  RMApp app = rm.submitApp(2000);

  //kick the scheduling
  nm1.nodeHeartbeat(true);

  RMAppAttempt attempt = app.getCurrentAppAttempt();
  MockAM am = rm.sendAMLaunched(attempt.getAppAttemptId());
  am.registerAppAttempt();
  am.unregisterAppAttempt();
  nm1.nodeHeartbeat(attempt.getAppAttemptId(), 1, ContainerState.COMPLETE);
  am.waitForState(RMAppAttemptState.FINISHED);
  rm.stop();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; //导入依赖的package包/类
public RMApp submitApp(int masterMemory, String name, String user,
    boolean unmanaged)
    throws Exception {
  return submitApp(masterMemory, name, user, null, unmanaged, null,
      super.getConfig().getInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS,
          YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS), null);
}