import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public void recover(RMState state) {
  ApplicationStateData 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.records.ApplicationStateData; //导入依赖的package包/类
@Override
protected void removeApplicationStateInternal(ApplicationStateData appState)
    throws IOException {
  ApplicationId appId =
      appState.getApplicationSubmissionContext().getApplicationId();
  String appKey = getApplicationNodeKey(appId);
  try {
    WriteBatch batch = db.createWriteBatch();
    try {
      batch.delete(bytes(appKey));
      for (ApplicationAttemptId attemptId : appState.attempts.keySet()) {
        String attemptKey = getApplicationAttemptNodeKey(appKey, attemptId);
        batch.delete(bytes(attemptKey));
      }
      if (LOG.isDebugEnabled()) {
        LOG.debug("Removing state for app " + appId + " and "
            + appState.attempts.size() + " attempts" + " at " + appKey);
      }
      db.write(batch);
    } finally {
      batch.close();
    }
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
private void loadApplicationAttemptState(ApplicationStateData appState,
    ApplicationId appId)
    throws Exception {
  String appPath = getNodePath(rmAppRoot, appId.toString());
  List<String> attempts = getChildrenWithRetries(appPath, false);
  for (String attemptIDStr : attempts) {
    if (attemptIDStr.startsWith(ApplicationAttemptId.appAttemptIdStrPrefix)) {
      String attemptPath = getNodePath(appPath, attemptIDStr);
      byte[] attemptData = getDataWithRetries(attemptPath, false);

      ApplicationAttemptStateDataPBImpl attemptState =
          new ApplicationAttemptStateDataPBImpl(
              ApplicationAttemptStateDataProto.parseFrom(attemptData));

      appState.attempts.put(attemptState.getAttemptId(), attemptState);
    }
  }
  LOG.debug("Done loading applications from ZK state store");
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public synchronized void updateApplicationStateInternal(ApplicationId appId,
    ApplicationStateData appStateDataPB) throws Exception {
  String nodeUpdatePath = getNodePath(rmAppRoot, appId.toString());

  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing final state info for app: " + appId + " at: "
        + nodeUpdatePath);
  }
  byte[] appStateData = appStateDataPB.getProto().toByteArray();

  if (existsWithRetries(nodeUpdatePath, false) != null) {
    setDataWithRetries(nodeUpdatePath, appStateData, -1);
  } else {
    createWithRetries(nodeUpdatePath, appStateData, zkAcl,
      CreateMode.PERSISTENT);
    LOG.debug(appId + " znode didn't exist. Created a new znode to"
        + " update the application state.");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public synchronized void removeApplicationStateInternal(
    ApplicationStateData  appState)
    throws Exception {
  String appId = appState.getApplicationSubmissionContext().getApplicationId()
      .toString();
  String appIdRemovePath = getNodePath(rmAppRoot, appId);
  ArrayList<Op> opList = new ArrayList<Op>();

  for (ApplicationAttemptId attemptId : appState.attempts.keySet()) {
    String attemptRemovePath = getNodePath(appIdRemovePath, attemptId.toString());
    opList.add(Op.delete(attemptRemovePath, -1));
  }
  opList.add(Op.delete(appIdRemovePath, -1));

  if (LOG.isDebugEnabled()) {
    LOG.debug("Removing info for app: " + appId + " at: " + appIdRemovePath
        + " and its attempts.");
  }
  doDeleteMultiWithRetries(opList);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public synchronized void storeApplicationStateInternal(ApplicationId appId,
    ApplicationStateData appStateDataPB) throws Exception {
  Path appDirPath = getAppDir(rmAppRoot, appId);
  mkdirsWithRetries(appDirPath);
  Path nodeCreatePath = getNodePath(appDirPath, appId.toString());

  LOG.info("Storing info for app: " + appId + " at: " + nodeCreatePath);
  byte[] appStateData = appStateDataPB.getProto().toByteArray();
  try {
    // currently throw all exceptions. May need to respond differently for HA
    // based on whether we have lost the right to write to FS
    writeFileWithRetries(nodeCreatePath, appStateData, true);
  } catch (Exception e) {
    LOG.info("Error storing info for app: " + appId, e);
    throw e;
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public synchronized void updateApplicationStateInternal(ApplicationId appId,
    ApplicationStateData appStateDataPB) throws Exception {
  Path appDirPath = getAppDir(rmAppRoot, appId);
  Path nodeCreatePath = getNodePath(appDirPath, appId.toString());

  LOG.info("Updating info for app: " + appId + " at: " + nodeCreatePath);
  byte[] appStateData = appStateDataPB.getProto().toByteArray();
  try {
    // currently throw all exceptions. May need to respond differently for HA
    // based on whether we have lost the right to write to FS
    updateFile(nodeCreatePath, appStateData, true);
  } catch (Exception e) {
    LOG.info("Error updating info for app: " + appId, e);
    throw e;
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public void transition(RMStateStore store, RMStateStoreEvent event) {
  if (!(event instanceof RMStateStoreAppEvent)) {
    // should never happen
    LOG.error("Illegal event type: " + event.getClass());
    return;
  }
  ApplicationStateData appState =
      ((RMStateStoreAppEvent) event).getAppState();
  ApplicationId appId =
      appState.getApplicationSubmissionContext().getApplicationId();
  LOG.info("Storing info for app: " + appId);
  try {
    store.storeApplicationStateInternal(appId, appState);
    store.notifyApplication(new RMAppEvent(appId,
           RMAppEventType.APP_NEW_SAVED));
  } catch (Exception e) {
    LOG.error("Error storing app: " + appId, e);
    store.notifyStoreOperationFailed(e);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的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.recovery.records.ApplicationStateData; //导入依赖的package包/类
protected RMApp testCreateAppSubmittedRecovery(
    ApplicationSubmissionContext submissionContext) throws IOException {
  RMApp application = createNewTestApp(submissionContext);
  // NEW => SUBMITTED event RMAppEventType.RECOVER
  RMState state = new RMState();
  ApplicationStateData appState =
      ApplicationStateData.newInstance(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.records.ApplicationStateData; //导入依赖的package包/类
@Override
public void recover(RMState state) {
  ApplicationStateData 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(null == appState.getDiagnostics() ? "" : appState
      .getDiagnostics());
  this.storedFinishTime = appState.getFinishTime();
  this.startTime = appState.getStartTime();
  this.callerContext = appState.getCallerContext();

  // send the ATS create Event
  sendATSCreateEvent(this, this.startTime);

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

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public synchronized void removeApplicationStateInternal(
    ApplicationStateData  appState)
    throws Exception {
  String appId = appState.getApplicationSubmissionContext().getApplicationId()
      .toString();
  String appIdRemovePath = getNodePath(rmAppRoot, appId);

  if (LOG.isDebugEnabled()) {
    LOG.debug("Removing info for app: " + appId + " at: " + appIdRemovePath
        + " and its attempts.");
  }

  for (ApplicationAttemptId attemptId : appState.attempts.keySet()) {
    String attemptRemovePath = getNodePath(appIdRemovePath, attemptId.toString());
    safeDelete(attemptRemovePath);
  }
  safeDelete(appIdRemovePath);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public RMStateStoreState transition(RMStateStore store,
    RMStateStoreEvent event) {
  if (!(event instanceof RMStateStoreAppEvent)) {
    // should never happen
    LOG.error("Illegal event type: " + event.getClass());
    return RMStateStoreState.ACTIVE;
  }
  boolean isFenced = false;
  ApplicationStateData appState =
      ((RMStateStoreAppEvent) event).getAppState();
  ApplicationId appId =
      appState.getApplicationSubmissionContext().getApplicationId();
  LOG.info("Storing info for app: " + appId);
  try {
    store.storeApplicationStateInternal(appId, appState);
    store.notifyApplication(new RMAppEvent(appId,
           RMAppEventType.APP_NEW_SAVED));
  } catch (Exception e) {
    LOG.error("Error storing app: " + appId, e);
    isFenced = store.notifyStoreOperationFailedInternal(e);
  }
  return finalState(isFenced);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的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;
  }
  ApplicationStateData appState =
      ApplicationStateData.newInstance(this.submitTime, this.startTime,
          this.user, this.submissionContext,
          stateToBeStored, diags, this.storedFinishTime);
  this.rmContext.getStateStore().updateApplicationState(appState);
}
 

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

  // create and recover app.
  RMAppImpl application =
      createAndPopulateNewRMApp(appContext, appState.getSubmitTime(),
          appState.getUser(), true);

  application.handle(new RMAppRecoverEvent(appId, rmState));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
private int loadRMApp(RMState rmState, LeveldbIterator iter, String appIdStr,
    byte[] appData) throws IOException {
  ApplicationStateData appState = createApplicationState(appIdStr, appData);
  ApplicationId appId =
      appState.getApplicationSubmissionContext().getApplicationId();
  rmState.appState.put(appId, appState);
  String attemptNodePrefix = getApplicationNodeKey(appId) + SEPARATOR;
  while (iter.hasNext()) {
    Entry<byte[],byte[]> entry = iter.peekNext();
    String key = asString(entry.getKey());
    if (!key.startsWith(attemptNodePrefix)) {
      break;
    }

    String attemptId = key.substring(attemptNodePrefix.length());
    if (attemptId.startsWith(ApplicationAttemptId.appAttemptIdStrPrefix)) {
      ApplicationAttemptStateData attemptState =
          createAttemptState(attemptId, entry.getValue());
      appState.attempts.put(attemptState.getAttemptId(), attemptState);
    } else {
      LOG.warn("Ignoring unknown application key: " + key);
    }
    iter.next();
  }
  int numAttempts = appState.attempts.size();
  if (LOG.isDebugEnabled()) {
    LOG.debug("Loaded application " + appId + " with " + numAttempts
        + " attempts");
  }
  return numAttempts;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
private ApplicationStateData createApplicationState(String appIdStr,
    byte[] data) throws IOException {
  ApplicationId appId = ConverterUtils.toApplicationId(appIdStr);
  ApplicationStateDataPBImpl appState =
      new ApplicationStateDataPBImpl(
          ApplicationStateDataProto.parseFrom(data));
  if (!appId.equals(
      appState.getApplicationSubmissionContext().getApplicationId())) {
    throw new YarnRuntimeException("The database entry for " + appId
        + " contains data for "
        + appState.getApplicationSubmissionContext().getApplicationId());
  }
  return appState;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@VisibleForTesting
ApplicationStateData loadRMAppState(ApplicationId appId) throws IOException {
  String appKey = getApplicationNodeKey(appId);
  byte[] data = null;
  try {
    data = db.get(bytes(appKey));
  } catch (DBException e) {
    throw new IOException(e);
  }
  if (data == null) {
    return null;
  }
  return createApplicationState(appId.toString(), data);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
protected void storeApplicationStateInternal(ApplicationId appId,
    ApplicationStateData appStateData) throws IOException {
  String key = getApplicationNodeKey(appId);
  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing state for app " + appId + " at " + key);
  }
  try {
    db.put(bytes(key), appStateData.getProto().toByteArray());
  } catch (DBException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; //导入依赖的package包/类
@Override
public synchronized void storeApplicationStateInternal(ApplicationId appId,
    ApplicationStateData appStateDataPB) throws Exception {
  String nodeCreatePath = getNodePath(rmAppRoot, appId.toString());

  if (LOG.isDebugEnabled()) {
    LOG.debug("Storing info for app: " + appId + " at: " + nodeCreatePath);
  }
  byte[] appStateData = appStateDataPB.getProto().toByteArray();
  createWithRetries(nodeCreatePath, appStateData, zkAcl,
    CreateMode.PERSISTENT);

}