import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@Override
public void recover(RMState state) {
  ApplicationState appState = state.getApplicationState().get(getApplicationId());
  this.recoveredFinalState = appState.getState();
  LOG.info("Recovering app: " + getApplicationId() + " with " + 
      + appState.getAttemptCount() + " attempts and final state = " + this.recoveredFinalState );
  this.diagnostics.append(appState.getDiagnostics());
  this.storedFinishTime = appState.getFinishTime();
  this.startTime = appState.getStartTime();

  for(int i=0; i<appState.getAttemptCount(); ++i) {
    // create attempt
    createNewAttempt();
    ((RMAppAttemptImpl)this.currentAttempt).recover(state);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的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, ApplicationState> 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
  ApplicationState 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.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@Override
public void recover(RMState state) throws Exception{
  ApplicationState appState = state.getApplicationState().get(getApplicationId());
  this.recoveredFinalState = appState.getState();
  LOG.info("Recovering app: " + getApplicationId() + " with " + 
      + appState.getAttemptCount() + " attempts and final state = " + this.recoveredFinalState );
  this.diagnostics.append(appState.getDiagnostics());
  this.storedFinishTime = appState.getFinishTime();
  this.startTime = appState.getStartTime();

  for(int i=0; i<appState.getAttemptCount(); ++i) {
    // create attempt
    createNewAttempt();
    ((RMAppAttemptImpl)this.currentAttempt).recover(state);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public void testRecoverApplication(ApplicationState appState, RMState rmState)
    throws Exception {
  ApplicationSubmissionContext submissionContext =
      appState.getApplicationSubmissionContext();
  RMAppImpl application =
      new RMAppImpl(appState.getAppId(), rmContext, conf,
          submissionContext.getApplicationName(), null,
          submissionContext.getQueue(), submissionContext, null, null,
          appState.getSubmitTime(), submissionContext.getApplicationType(),
          submissionContext.getApplicationTags());
  Assert.assertEquals(RMAppState.NEW, application.getState());
  application.recover(rmState);

  // Application final status looked from recoveredFinalStatus
  Assert.assertTrue("Application is not in recoveredFinalStatus.",
      RMAppImpl.isAppInFinalState(application));

  // Trigger RECOVER event.
  application.handle(new RMAppEvent(appState.getAppId(),
      RMAppEventType.RECOVER));
  rmDispatcher.await();
  RMAppState finalState = appState.getState();
  Assert.assertEquals("Application is not in finalState.", finalState,
      application.getState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
private void rememberTargetTransitionsAndStoreState(RMAppEvent event,
    Object transitionToDo, RMAppState targetFinalState,
    RMAppState stateToBeStored) {
  rememberTargetTransitions(event, transitionToDo, targetFinalState);
  this.stateBeforeFinalSaving = getState();
  this.storedFinishTime = this.systemClock.getTime();

  LOG.info("Updating application " + this.applicationId
      + " with final state: " + this.targetedFinalState);
  // we lost attempt_finished diagnostics in app, because attempt_finished
  // diagnostics is sent after app final state is saved. Later on, we will
  // create GetApplicationAttemptReport specifically for getting per attempt
  // info.
  String diags = null;
  switch (event.getType()) {
  case APP_REJECTED:
    RMAppRejectedEvent rejectedEvent = (RMAppRejectedEvent) event;
    diags = rejectedEvent.getMessage();
    break;
  case ATTEMPT_FINISHED:
    RMAppFinishedAttemptEvent finishedEvent =
        (RMAppFinishedAttemptEvent) event;
    diags = finishedEvent.getDiagnostics();
    break;
  case ATTEMPT_FAILED:
    RMAppFailedAttemptEvent failedEvent = (RMAppFailedAttemptEvent) event;
    diags = getAppAttemptFailedDiagnostics(failedEvent);
    break;
  case ATTEMPT_KILLED:
    diags = getAppKilledDiagnostics();
    break;
  default:
    break;
  }
  ApplicationState appState =
      new ApplicationState(this.submitTime, this.startTime,
        this.submissionContext, this.user, stateToBeStored, diags,
        this.storedFinishTime);
  this.rmContext.getStateStore().updateApplicationState(appState);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
protected void recoverApplication(ApplicationState appState, RMState rmState)
    throws Exception {
  ApplicationSubmissionContext appContext =
      appState.getApplicationSubmissionContext();
  ApplicationId appId = appState.getAppId();

  // create and recover app.
  RMAppImpl application =
      createAndPopulateNewRMApp(appContext, appState.getSubmitTime(),
        appState.getUser());
  application.handle(new RMAppRecoverEvent(appId, rmState));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public static ApplicationStateData newInstance(
    ApplicationState appState) {
  return newInstance(appState.getSubmitTime(), appState.getStartTime(),
      appState.getUser(), appState.getApplicationSubmissionContext(),
      appState.getState(), appState.getDiagnostics(),
      appState.getFinishTime());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
protected RMApp testCreateAppSubmittedRecovery(
    ApplicationSubmissionContext submissionContext) throws IOException {
  RMApp application = createNewTestApp(submissionContext);
  // NEW => SUBMITTED event RMAppEventType.RECOVER
  RMState state = new RMState();
  ApplicationState appState = new ApplicationState(123, 123, null, "user");
  state.getApplicationState().put(application.getApplicationId(), appState);
  RMAppEvent event =
      new RMAppRecoverEvent(application.getApplicationId(), state);

  application.handle(event);
  assertStartTimeSet(application);
  assertAppState(RMAppState.SUBMITTED, application);
  return application;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@Test(timeout = 30000)
public void testAppsRecoveringStates() throws Exception {
  RMState state = new RMState();
  Map<ApplicationId, ApplicationState> applicationState =
      state.getApplicationState();
  createRMStateForApplications(applicationState, RMAppState.FINISHED);
  createRMStateForApplications(applicationState, RMAppState.KILLED);
  createRMStateForApplications(applicationState, RMAppState.FAILED);
  for (ApplicationState appState : applicationState.values()) {
    testRecoverApplication(appState, state);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public void testRecoverApplication(ApplicationState appState, RMState rmState)
    throws Exception {
  ApplicationSubmissionContext submissionContext =
      appState.getApplicationSubmissionContext();
  RMAppImpl application =
      new RMAppImpl(appState.getAppId(), rmContext, conf,
          submissionContext.getApplicationName(), null,
          submissionContext.getQueue(), submissionContext, null, null,
          appState.getSubmitTime(), submissionContext.getApplicationType(),
          submissionContext.getApplicationTags(),
          BuilderUtils.newResourceRequest(
              RMAppAttemptImpl.AM_CONTAINER_PRIORITY, ResourceRequest.ANY,
              submissionContext.getResource(), 1));
  Assert.assertEquals(RMAppState.NEW, application.getState());

  RMAppEvent recoverEvent =
      new RMAppRecoverEvent(application.getApplicationId(), rmState);
  // Trigger RECOVER event.
  application.handle(recoverEvent);
  // Application final status looked from recoveredFinalStatus
  Assert.assertTrue("Application is not in recoveredFinalStatus.",
      RMAppImpl.isAppInFinalState(application));

  rmDispatcher.await();
  RMAppState finalState = appState.getState();
  Assert.assertEquals("Application is not in finalState.", finalState,
      application.getState());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public void createRMStateForApplications(
    Map<ApplicationId, ApplicationState> applicationState,
    RMAppState rmAppState) {
  RMApp app = createNewTestApp(null);
  ApplicationState appState =
      new ApplicationState(app.getSubmitTime(), app.getStartTime(),
          app.getApplicationSubmissionContext(), app.getUser(), rmAppState,
          null, app.getFinishTime());
  applicationState.put(app.getApplicationId(), appState);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@Override
public void recover(RMState state) {
  ApplicationState appState = state.getApplicationState().get(getApplicationId());
  LOG.info("Recovering app: " + getApplicationId() + " with " + 
          + appState.getAttemptCount() + " attempts");
  for(int i=0; i<appState.getAttemptCount(); ++i) {
    // create attempt
    createNewAttempt(false);
    // recover attempt
    ((RMAppAttemptImpl) currentAttempt).recover(state);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@Override
public void recover(RMState state) throws Exception{
  ApplicationState appState = state.getApplicationState().get(getApplicationId());
  LOG.info("Recovering app: " + getApplicationId() + " with " + 
          + appState.getAttemptCount() + " attempts");
  for(int i=0; i<appState.getAttemptCount(); ++i) {
    // create attempt
    createNewAttempt(false);
    // recover attempt
    ((RMAppAttemptImpl) currentAttempt).recover(state);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
private void rememberTargetTransitionsAndStoreState(RMAppEvent event,
    Object transitionToDo, RMAppState targetFinalState,
    RMAppState stateToBeStored) {
  rememberTargetTransitions(event, transitionToDo, targetFinalState);
  this.stateBeforeFinalSaving = getState();
  this.storedFinishTime = System.currentTimeMillis();

  LOG.info("Updating application " + this.applicationId
      + " with final state: " + this.targetedFinalState);
  // we lost attempt_finished diagnostics in app, because attempt_finished
  // diagnostics is sent after app final state is saved. Later on, we will
  // create GetApplicationAttemptReport specifically for getting per attempt
  // info.
  String diags = null;
  switch (event.getType()) {
  case APP_REJECTED:
    RMAppRejectedEvent rejectedEvent = (RMAppRejectedEvent) event;
    diags = rejectedEvent.getMessage();
    break;
  case ATTEMPT_FINISHED:
    RMAppFinishedAttemptEvent finishedEvent =
        (RMAppFinishedAttemptEvent) event;
    diags = finishedEvent.getDiagnostics();
    break;
  case ATTEMPT_FAILED:
    RMAppFailedAttemptEvent failedEvent = (RMAppFailedAttemptEvent) event;
    diags = getAppAttemptFailedDiagnostics(failedEvent);
    break;
  case ATTEMPT_KILLED:
    diags = getAppKilledDiagnostics();
    break;
  default:
    break;
  }
  ApplicationState appState =
      new ApplicationState(this.submitTime, this.startTime,
        this.submissionContext, this.user, stateToBeStored, diags,
        this.storedFinishTime);
  this.rmContext.getStateStore().updateApplicationState(appState);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@SuppressWarnings("unchecked")
protected void
    recoverApplication(ApplicationState appState, RMState rmState)
        throws Exception {
  ApplicationSubmissionContext appContext =
      appState.getApplicationSubmissionContext();
  ApplicationId appId = appState.getAppId();

  // create and recover app.
  RMAppImpl application =
      createAndPopulateNewRMApp(appContext, appState.getSubmitTime(),
        appState.getUser());
  application.recover(rmState);
  if (isApplicationInFinalState(appState.getState())) {
    // We are synchronously moving the application into final state so that
    // momentarily client will not see this application in NEW state. Also
    // for finished applications we will avoid renewing tokens.
    application.handle(new RMAppEvent(appId, RMAppEventType.RECOVER));
    return;
  }

  if (UserGroupInformation.isSecurityEnabled()) {
    Credentials credentials = null;
    try {
      credentials = parseCredentials(appContext);
      // synchronously renew delegation token on recovery.
      rmContext.getDelegationTokenRenewer().addApplicationSync(appId,
        credentials, appContext.getCancelTokensWhenComplete());
      application.handle(new RMAppEvent(appId, RMAppEventType.RECOVER));
    } catch (Exception e) {
      LOG.warn("Unable to parse and renew delegation tokens.", e);
      this.rmContext.getDispatcher().getEventHandler()
        .handle(new RMAppRejectedEvent(appId, e.getMessage()));
      throw e;
    }
  } else {
    application.handle(new RMAppEvent(appId, RMAppEventType.RECOVER));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public RMStateStoreAppEvent(ApplicationState appState) {
  super(RMStateStoreEventType.STORE_APP);
  this.appState = appState;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public ApplicationState getAppState() {
  return appState;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
RMStateStoreRemoveAppEvent(ApplicationState appState) {
  super(RMStateStoreEventType.REMOVE_APP);
  this.appState = appState;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
public RMStateUpdateAppEvent(ApplicationState appState) {
  super(RMStateStoreEventType.UPDATE_APP);
  this.appState = appState;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; //导入依赖的package包/类
@Test (timeout = 60000)
public void testRMRestartAppRunningAMFailed() throws Exception {
  conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS,
    YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS);
  MemoryRMStateStore memStore = new MemoryRMStateStore();
  memStore.init(conf);
  RMState rmState = memStore.getState();
  Map<ApplicationId, ApplicationState> rmAppState =
      rmState.getApplicationState();

  // start RM
  MockRM rm1 = createMockRM(conf, memStore);
  rm1.start();
  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, "name", "user",
        new HashMap<ApplicationAccessType, String>(), false, "default", -1,
        null, "MAPREDUCE", true, true);
  MockAM am0 = launchAM(app0, rm1, nm1);

  // fail the AM by sending CONTAINER_FINISHED event without registering.
  nm1.nodeHeartbeat(am0.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
  am0.waitForState(RMAppAttemptState.FAILED);

  ApplicationState appState = rmAppState.get(app0.getApplicationId());
  // assert the AM failed state is saved.
  Assert.assertEquals(RMAppAttemptState.FAILED,
    appState.getAttempt(am0.getApplicationAttemptId()).getState());

  // assert app state has not been saved.
  Assert.assertNull(rmAppState.get(app0.getApplicationId()).getState());

  // new AM started but not registered, app still stays at ACCECPTED state.
  rm1.waitForState(app0.getApplicationId(), RMAppState.ACCEPTED);

  // start new RM
  MockRM rm2 = createMockRM(conf, memStore);
  rm2.start();
  // assert the previous AM state is loaded back on RM recovery.

  rm2.waitForState(am0.getApplicationAttemptId(), RMAppAttemptState.FAILED);
}