import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的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.attempt.RMAppAttemptState; //导入依赖的package包/类
synchronized public void stop(RMAppAttemptState rmAppAttemptFinalState) {
  // clear pending resources metrics for the application
  QueueMetrics metrics = queue.getMetrics();
  for (Map<String, ResourceRequest> asks : requests.values()) {
    ResourceRequest request = asks.get(ResourceRequest.ANY);
    if (request != null) {
      metrics.decrPendingResources(user, request.getNumContainers(),
          request.getCapability());
      
      // Update Queue
      queue.decPendingResource(
          request.getNodeLabelExpression(),
          Resources.multiply(request.getCapability(),
              request.getNumContainers()));
    }
  }
  metrics.finishAppAttempt(applicationId, pending, user);
  
  // Clear requests themselves
  clearRequests();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public static ApplicationAttemptStateData newInstance(
    ApplicationAttemptId attemptId, Container container,
    Credentials attemptTokens, long startTime, RMAppAttemptState finalState,
    String finalTrackingUrl, String diagnostics,
    FinalApplicationStatus amUnregisteredFinalStatus, int exitStatus,
    long finishTime, long memorySeconds, long vcoreSeconds, long gcoreSeconds) {
  ApplicationAttemptStateData attemptStateData =
      Records.newRecord(ApplicationAttemptStateData.class);
  attemptStateData.setAttemptId(attemptId);
  attemptStateData.setMasterContainer(container);
  attemptStateData.setAppAttemptTokens(attemptTokens);
  attemptStateData.setState(finalState);
  attemptStateData.setFinalTrackingUrl(finalTrackingUrl);
  attemptStateData.setDiagnostics(diagnostics == null ? "" : diagnostics);
  attemptStateData.setStartTime(startTime);
  attemptStateData.setFinalApplicationStatus(amUnregisteredFinalStatus);
  attemptStateData.setAMContainerExitStatus(exitStatus);
  attemptStateData.setFinishTime(finishTime);
  attemptStateData.setMemorySeconds(memorySeconds);
  attemptStateData.setVcoreSeconds(vcoreSeconds);
  attemptStateData.setGcoreSeconds(gcoreSeconds);
  return attemptStateData;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的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.attempt.RMAppAttemptState; //导入依赖的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.attempt.RMAppAttemptState; //导入依赖的package包/类
private void finishApplicationMaster(RMApp rmApp, MockRM rm, MockNM nm,
    MockAM am, FinishApplicationMasterRequest req) throws Exception {
  RMState rmState =
      ((MemoryRMStateStore) rm.getRMContext().getStateStore()).getState();
  Map<ApplicationId, ApplicationStateData> rmAppState =
      rmState.getApplicationState();
  am.unregisterAppAttempt(req,true);
  am.waitForState(RMAppAttemptState.FINISHING);
  nm.nodeHeartbeat(am.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
  am.waitForState(RMAppAttemptState.FINISHED);
  rm.waitForState(rmApp.getApplicationId(), RMAppState.FINISHED);
  // check that app/attempt is saved with the final state
  ApplicationStateData appState = rmAppState.get(rmApp.getApplicationId());
  Assert
    .assertEquals(RMAppState.FINISHED, appState.getState());
  Assert.assertEquals(RMAppAttemptState.FINISHED,
    appState.getAttempt(am.getApplicationAttemptId()).getState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public void waitForState(ApplicationAttemptId attemptId, 
                         RMAppAttemptState finalState)
    throws Exception {
  RMApp app = getRMContext().getRMApps().get(attemptId.getApplicationId());
  Assert.assertNotNull("app shouldn't be null", app);
  RMAppAttempt attempt = app.getRMAppAttempt(attemptId);
  int timeoutSecs = 0;
  while (!finalState.equals(attempt.getAppAttemptState()) && timeoutSecs++ < 40) {
    System.out.println("AppAttempt : " + attemptId 
        + " State is : " + attempt.getAppAttemptState()
        + " Waiting for state : " + finalState);
    Thread.sleep(1000);
  }
  System.out.println("Attempt State is : " + attempt.getAppAttemptState());
  Assert.assertEquals("Attempt state is not correct (timedout)", finalState,
      attempt.getAppAttemptState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public MockAM sendAMLaunched(ApplicationAttemptId appAttemptId)
    throws Exception {
  MockAM am = new MockAM(getRMContext(), masterService, appAttemptId);
  am.waitForState(RMAppAttemptState.ALLOCATED);
  //create and set AMRMToken
  Token<AMRMTokenIdentifier> amrmToken =
      this.rmContext.getAMRMTokenSecretManager().createAndGetAMRMToken(
        appAttemptId);
  ((RMAppAttemptImpl) this.rmContext.getRMApps()
    .get(appAttemptId.getApplicationId()).getRMAppAttempt(appAttemptId))
    .setAMRMToken(amrmToken);
  getRMContext()
      .getDispatcher()
      .getEventHandler()
      .handle(
          new RMAppAttemptEvent(appAttemptId, RMAppAttemptEventType.LAUNCHED));
  return am;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public void waitForState(RMAppAttemptState finalState) throws Exception {
  RMApp app = context.getRMApps().get(attemptId.getApplicationId());
  RMAppAttempt attempt = app.getRMAppAttempt(attemptId);
  int timeoutSecs = 0;
  while (!finalState.equals(attempt.getAppAttemptState())
      && timeoutSecs++ < 40) {
    System.out
        .println("AppAttempt : " + attemptId + " State is : " 
            + attempt.getAppAttemptState()
            + " Waiting for state : " + finalState);
    Thread.sleep(1000);
  }
  System.out.println("AppAttempt State is : " + attempt.getAppAttemptState());
  Assert.assertEquals("AppAttempt state is not correct (timedout)",
      finalState, attempt.getAppAttemptState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public void unregisterAppAttempt(final FinishApplicationMasterRequest req,
    boolean waitForStateRunning) throws Exception {
  if (waitForStateRunning) {
    waitForState(RMAppAttemptState.RUNNING);
  }
  if (ugi == null) {
    ugi =  UserGroupInformation.createRemoteUser(attemptId.toString());
    Token<AMRMTokenIdentifier> token =
        context.getRMApps()
            .get(attemptId.getApplicationId())
            .getRMAppAttempt(attemptId).getAMRMToken();
    ugi.addTokenIdentifier(token.decodeIdentifier());
  }
  try {
    ugi.doAs(new PrivilegedExceptionAction<Object>() {
      @Override
      public Object run() throws Exception {
        amRMProtocol.finishApplicationMaster(req);
        return null;
      }
    });
  } catch (UndeclaredThrowableException e) {
    throw (Exception) e.getCause();
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的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.attempt.RMAppAttemptState; //导入依赖的package包/类
private void checkActiveRMFunctionality() throws Exception {
  assertEquals(STATE_ERR, HAServiceState.ACTIVE,
      rm.adminService.getServiceStatus().getState());
  assertTrue("Active RM services aren't started",
      rm.areActiveServicesRunning());
  assertTrue("RM is not ready to become active",
      rm.adminService.getServiceStatus().isReadyToBecomeActive());

  try {
    rm.getNewAppId();
    rm.registerNode("127.0.0.1:1", 2048);
    app = rm.submitApp(1024);
    attempt = app.getCurrentAppAttempt();
    rm.waitForState(attempt.getAppAttemptId(), RMAppAttemptState.SCHEDULED);
  } catch (Exception e) {
    fail("Unable to perform Active RM functions");
    LOG.error("ActiveRM check failed", e);
  }

  checkActiveRMWebServices();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public void appAttemptFinished(RMAppAttempt appAttempt,
    RMAppAttemptState appAttemtpState, RMApp app, long finishedTime) {
  if (publishSystemMetrics) {
    ContainerId container = (appAttempt.getMasterContainer() == null) ? null
        : appAttempt.getMasterContainer().getId();
    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,
            container));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
synchronized public void stop(RMAppAttemptState rmAppAttemptFinalState) {
  // clear pending resources metrics for the application
  QueueMetrics metrics = queue.getMetrics();
  for (Map<String, ResourceRequest> asks : resourceRequestMap.values()) {
    ResourceRequest request = asks.get(ResourceRequest.ANY);
    if (request != null) {
      metrics.decrPendingResources(user, request.getNumContainers(),
          request.getCapability());
      
      // Update Queue
      queue.decPendingResource(
          request.getNodeLabelExpression(),
          Resources.multiply(request.getCapability(),
              request.getNumContainers()));
    }
  }
  metrics.finishAppAttempt(applicationId, pending, user);
  
  // Clear requests themselves
  clearRequests();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public static ApplicationAttemptStateData newInstance(
    ApplicationAttemptId attemptId, Container container,
    Credentials attemptTokens, long startTime, RMAppAttemptState finalState,
    String finalTrackingUrl, String diagnostics,
    FinalApplicationStatus amUnregisteredFinalStatus, int exitStatus,
    long finishTime, long memorySeconds, long vcoreSeconds) {
  ApplicationAttemptStateData attemptStateData =
      Records.newRecord(ApplicationAttemptStateData.class);
  attemptStateData.setAttemptId(attemptId);
  attemptStateData.setMasterContainer(container);
  attemptStateData.setAppAttemptTokens(attemptTokens);
  attemptStateData.setState(finalState);
  attemptStateData.setFinalTrackingUrl(finalTrackingUrl);
  attemptStateData.setDiagnostics(diagnostics == null ? "" : diagnostics);
  attemptStateData.setStartTime(startTime);
  attemptStateData.setFinalApplicationStatus(amUnregisteredFinalStatus);
  attemptStateData.setAMContainerExitStatus(exitStatus);
  attemptStateData.setFinishTime(finishTime);
  attemptStateData.setMemorySeconds(memorySeconds);
  attemptStateData.setVcoreSeconds(vcoreSeconds);
  return attemptStateData;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
@Test(timeout = 10000)
public void testPublishAppAttemptMetricsForUnmanagedAM() throws Exception {
  ApplicationAttemptId appAttemptId =
      ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1);
  RMAppAttempt appAttempt = createRMAppAttempt(appAttemptId,true);
  metricsPublisher.appAttemptRegistered(appAttempt, Integer.MAX_VALUE + 1L);
  RMApp app = mock(RMApp.class);
  when(app.getFinalApplicationStatus()).thenReturn(FinalApplicationStatus.UNDEFINED);
  metricsPublisher.appAttemptFinished(appAttempt, RMAppAttemptState.FINISHED, app,
      Integer.MAX_VALUE + 2L);
  TimelineEntity entity = null;
  do {
    entity =
        store.getEntity(appAttemptId.toString(),
            AppAttemptMetricsConstants.ENTITY_TYPE,
            EnumSet.allOf(Field.class));
    // ensure two events are both published before leaving the loop
  } while (entity == null || entity.getEvents().size() < 2);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public void waitForState(RMAppAttemptState finalState) throws Exception {
  RMApp app = context.getRMApps().get(attemptId.getApplicationId());
  RMAppAttempt attempt = app.getRMAppAttempt(attemptId);
  final int timeoutMsecs = 40000;
  final int minWaitMsecs = 1000;
  final int waitMsPerLoop = 500;
  int loop = 0;
  while (!finalState.equals(attempt.getAppAttemptState())
      && waitMsPerLoop * loop < timeoutMsecs) {
    LOG.info("AppAttempt : " + attemptId + " State is : " +
        attempt.getAppAttemptState() + " Waiting for state : " +
        finalState);
    Thread.yield();
    Thread.sleep(waitMsPerLoop);
    loop++;
  }
  int waitedMsecs = waitMsPerLoop * loop;
  if (minWaitMsecs > waitedMsecs) {
    Thread.sleep(minWaitMsecs - waitedMsecs);
  }
  LOG.info("Attempt State is : " + attempt.getAppAttemptState());
  if (waitedMsecs >= timeoutMsecs) {
    Assert.fail("Attempt state is not correct (timedout): expected: "
        + finalState + " actual: " + attempt.getAppAttemptState());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
public static YarnApplicationAttemptState createApplicationAttemptState(
    RMAppAttemptState rmAppAttemptState) {
  switch (rmAppAttemptState) {
    case NEW:
      return YarnApplicationAttemptState.NEW;
    case SUBMITTED:
      return YarnApplicationAttemptState.SUBMITTED;
    case SCHEDULED:
      return YarnApplicationAttemptState.SCHEDULED;
    case ALLOCATED:
      return YarnApplicationAttemptState.ALLOCATED;
    case LAUNCHED:
      return YarnApplicationAttemptState.LAUNCHED;
    case ALLOCATED_SAVING:
    case LAUNCHED_UNMANAGED_SAVING:
      return YarnApplicationAttemptState.ALLOCATED_SAVING;
    case RUNNING:
      return YarnApplicationAttemptState.RUNNING;
    case FINISHING:
      return YarnApplicationAttemptState.FINISHING;
    case FINISHED:
      return YarnApplicationAttemptState.FINISHED;
    case KILLED:
      return YarnApplicationAttemptState.KILLED;
    case FAILED:
      return YarnApplicationAttemptState.FAILED;
    default:
      throw new YarnRuntimeException("Unknown state passed!");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public void applicationAttemptFinished(RMAppAttempt appAttempt,
    RMAppAttemptState finalState) {
  if (historyServiceEnabled) {
    dispatcher.getEventHandler().handle(
      new WritingApplicationAttemptFinishEvent(appAttempt.getAppAttemptId(),
        ApplicationAttemptFinishData.newInstance(
          appAttempt.getAppAttemptId(), appAttempt.getDiagnostics()
            .toString(), appAttempt.getTrackingUrl(), appAttempt
            .getFinalApplicationStatus(),
            RMServerUtils.createApplicationAttemptState(finalState))));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; //导入依赖的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);
}