import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void handleDTRenewerAppSubmitEvent(
    DelegationTokenRenewerAppSubmitEvent event) {
  /*
   * For applications submitted with delegation tokens we are not submitting
   * the application to scheduler from RMAppManager. Instead we are doing
   * it from here. The primary goal is to make token renewal as a part of
   * application submission asynchronous so that client thread is not
   * blocked during app submission.
   */
  try {
    // Setup tokens for renewal
    DelegationTokenRenewer.this.handleAppSubmitEvent(event);
    rmContext.getDispatcher().getEventHandler()
        .handle(new RMAppEvent(event.getApplicationId(), RMAppEventType.START));
  } catch (Throwable t) {
    LOG.warn(
        "Unable to add the application to the delegation token renewer.",
        t);
    // Sending APP_REJECTED is fine, since we assume that the
    // RMApp is in NEW state and thus we havne't yet informed the
    // Scheduler about the existence of the application
    rmContext.getDispatcher().getEventHandler().handle(
        new RMAppRejectedEvent(event.getApplicationId(), t.getMessage()));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * {@link RMAppAttemptState#SUBMITTED} -> {@link RMAppAttemptState#FAILED}
 */
private void testAppAttemptSubmittedToFailedState(String diagnostics) {
  sendAttemptUpdateSavedEvent(applicationAttempt);
  assertEquals(RMAppAttemptState.FAILED, 
      applicationAttempt.getAppAttemptState());
  assertEquals(diagnostics, applicationAttempt.getDiagnostics());
  assertEquals(0,applicationAttempt.getJustFinishedContainers().size());
  assertNull(applicationAttempt.getMasterContainer());
  assertEquals(0.0, (double)applicationAttempt.getProgress(), 0.0001);
  assertEquals(0, application.getRanNodes().size());
  assertNull(applicationAttempt.getFinalApplicationStatus());
  
  // Check events
  verify(masterService).
      unregisterAttempt(applicationAttempt.getAppAttemptId());
  
  // this works for unmanaged and managed AM's because this is actually doing
  // verify(application).handle(anyObject());
  verify(application).handle(any(RMAppRejectedEvent.class));
  verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
  verifyApplicationAttemptFinished(RMAppAttemptState.FAILED);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
@Override
public void transition(RMAppAttemptImpl appAttempt,
    RMAppAttemptEvent event) {

  RMAppAttemptRejectedEvent rejectedEvent = (RMAppAttemptRejectedEvent) event;

  // Tell the AMS. Unregister from the ApplicationMasterService
  appAttempt.masterService
      .unregisterAttempt(appAttempt.applicationAttemptId);
  
  // Save the diagnostic message
  String message = rejectedEvent.getMessage();
  appAttempt.setDiagnostics(message);

  // Send the rejection event to app
  appAttempt.eventHandler.handle(
      new RMAppRejectedEvent(
          rejectedEvent.getApplicationAttemptId().getApplicationId(),
          message)
      );
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * {@link RMAppAttemptState#SUBMITTED} -> {@link RMAppAttemptState#FAILED}
 */
private void testAppAttemptSubmittedToFailedState(String diagnostics) {
  assertEquals(RMAppAttemptState.FAILED, 
      applicationAttempt.getAppAttemptState());
  assertEquals(diagnostics, applicationAttempt.getDiagnostics());
  assertEquals(0,applicationAttempt.getJustFinishedContainers().size());
  assertNull(applicationAttempt.getMasterContainer());
  assertEquals(0.0, (double)applicationAttempt.getProgress(), 0.0001);
  assertEquals(0, applicationAttempt.getRanNodes().size());
  assertNull(applicationAttempt.getFinalApplicationStatus());
  
  // Check events
  verify(masterService).
      unregisterAttempt(applicationAttempt.getAppAttemptId());
  
  // this works for unmanaged and managed AM's because this is actually doing
  // verify(application).handle(anyObject());
  verify(application).handle(any(RMAppRejectedEvent.class));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
@Override
public void transition(RMAppAttemptImpl appAttempt,
    RMAppAttemptEvent event) {

  RMAppAttemptRejectedEvent rejectedEvent = (RMAppAttemptRejectedEvent) event;

  // Tell the AMS. Unregister from the ApplicationMasterService
  appAttempt.masterService
      .unregisterAttempt(appAttempt.applicationAttemptId);
  
  // Save the diagnostic message
  String message = rejectedEvent.getMessage();
  appAttempt.setDiagnostics(message);

  // Send the rejection event to app
  appAttempt.eventHandler.handle(
      new RMAppRejectedEvent(
          rejectedEvent.getApplicationAttemptId().getApplicationId(),
          message)
      );

  appAttempt.removeCredentials(appAttempt);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * {@link RMAppAttemptState#SUBMITTED} -> {@link RMAppAttemptState#FAILED}
 */
private void testAppAttemptSubmittedToFailedState(String diagnostics) {
  assertEquals(RMAppAttemptState.FAILED, 
      applicationAttempt.getAppAttemptState());
  assertEquals(diagnostics, applicationAttempt.getDiagnostics());
  assertEquals(0,applicationAttempt.getJustFinishedContainers().size());
  assertNull(applicationAttempt.getMasterContainer());
  assertEquals(0.0, (double)applicationAttempt.getProgress(), 0.0001);
  assertEquals(0, applicationAttempt.getRanNodes().size());
  assertNull(applicationAttempt.getFinalApplicationStatus());
  
  // Check events
  verify(masterService).
      unregisterAttempt(applicationAttempt.getAppAttemptId());
  
  // this works for unmanaged and managed AM's because this is actually doing
  // verify(application).handle(anyObject());
  verify(application).handle(any(RMAppRejectedEvent.class));
  verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * {@link RMAppAttemptState#SUBMITTED} -> {@link RMAppAttemptState#FAILED}
 */
private void testAppAttemptSubmittedToFailedState(String diagnostics) {
  sendAttemptUpdateSavedEvent(applicationAttempt);
  assertEquals(RMAppAttemptState.FAILED, 
      applicationAttempt.getAppAttemptState());
  assertEquals(diagnostics, applicationAttempt.getDiagnostics());
  assertEquals(0,applicationAttempt.getJustFinishedContainers().size());
  assertNull(applicationAttempt.getMasterContainer());
  assertEquals(0.0, (double)applicationAttempt.getProgress(), 0.0001);
  assertEquals(0, applicationAttempt.getRanNodes().size());
  assertNull(applicationAttempt.getFinalApplicationStatus());
  
  // Check events
  verify(masterService).
      unregisterAttempt(applicationAttempt.getAppAttemptId());
  
  // this works for unmanaged and managed AM's because this is actually doing
  // verify(application).handle(anyObject());
  verify(application).handle(any(RMAppRejectedEvent.class));
  verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
  verifyApplicationAttemptFinished(RMAppAttemptState.FAILED);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * Helper method that attempts to assign the app to a queue. The method is
 * responsible to call the appropriate event-handler if the app is rejected.
 */
@VisibleForTesting
FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) {
  FSLeafQueue queue = null;
  String appRejectMsg = null;

  try {
    QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy();
    queueName = placementPolicy.assignAppToQueue(queueName, user);
    if (queueName == null) {
      appRejectMsg = "Application rejected by queue placement policy";
    } else {
      queue = queueMgr.getLeafQueue(queueName, true);
      if (queue == null) {
        appRejectMsg = queueName + " is not a leaf queue";
      }
    }
  } catch (IOException ioe) {
    appRejectMsg = "Error assigning app to queue " + queueName;
  }

  if (appRejectMsg != null && rmApp != null) {
    LOG.error(appRejectMsg);
    rmContext.getDispatcher().getEventHandler().handle(
        new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg));
    return null;
  }

  if (rmApp != null) {
    rmApp.setQueue(queue.getName());
  } else {
    LOG.error("Couldn't find RM app to set queue name on");
  }
  return queue;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * Helper method that attempts to assign the app to a queue. The method is
 * responsible to call the appropriate event-handler if the app is rejected.
 */
@VisibleForTesting
FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) {
  FSLeafQueue queue = null;
  String appRejectMsg = null;

  try {
    QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy();
    queueName = placementPolicy.assignAppToQueue(queueName, user);
    if (queueName == null) {
      appRejectMsg = "Application rejected by queue placement policy";
    } else {
      queue = queueMgr.getLeafQueue(queueName, true);
      if (queue == null) {
        appRejectMsg = queueName + " is not a leaf queue";
      }
    }
  } catch (InvalidQueueNameException qne) {
    appRejectMsg = qne.getMessage();
  } catch (IOException ioe) {
    appRejectMsg = "Error assigning app to queue " + queueName;
  }

  if (appRejectMsg != null && rmApp != null) {
    LOG.error(appRejectMsg);
    rmContext.getDispatcher().getEventHandler().handle(
        new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg));
    return null;
  }

  if (rmApp != null) {
    rmApp.setQueue(queue.getName());
  } else {
    LOG.error("Couldn't find RM app to set queue name on");
  }
  return queue;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的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.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
@SuppressWarnings("unchecked")
protected void submitApplication(
    ApplicationSubmissionContext submissionContext, long submitTime,
    boolean isRecovered, String user) throws YarnException {
  ApplicationId applicationId = submissionContext.getApplicationId();

  // Validation of the ApplicationSubmissionContext needs to be completed
  // here. Only those fields that are dependent on RM's configuration are
  // checked here as they have to be validated whether they are part of new
  // submission or just being recovered.

  // Check whether AM resource requirements are within required limits
  if (!submissionContext.getUnmanagedAM()) {
    ResourceRequest amReq = BuilderUtils.newResourceRequest(
        RMAppAttemptImpl.AM_CONTAINER_PRIORITY, ResourceRequest.ANY,
        submissionContext.getResource(), 1);
    try {
      SchedulerUtils.validateResourceRequest(amReq,
          scheduler.getMaximumResourceCapability());
    } catch (InvalidResourceRequestException e) {
      LOG.warn("RM app submission failed in validating AM resource request"
          + " for application " + applicationId, e);
      throw e;
    }
  }

  // Create RMApp
  RMApp application =
      new RMAppImpl(applicationId, rmContext, this.conf,
          submissionContext.getApplicationName(), user,
          submissionContext.getQueue(),
          submissionContext, this.scheduler, this.masterService,
          submitTime, submissionContext.getApplicationType());

  // Concurrent app submissions with same applicationId will fail here
  // Concurrent app submissions with different applicationIds will not
  // influence each other
  if (rmContext.getRMApps().putIfAbsent(applicationId, application) !=
      null) {
    String message = "Application with id " + applicationId
        + " is already present! Cannot add a duplicate!";
    LOG.warn(message);
    throw RPCUtil.getRemoteException(message);
  }

  // Inform the ACLs Manager
  this.applicationACLsManager.addApplication(applicationId,
      submissionContext.getAMContainerSpec().getApplicationACLs());

  try {
    // Setup tokens for renewal
    if (UserGroupInformation.isSecurityEnabled()) {
      this.rmContext.getDelegationTokenRenewer().addApplication(
          applicationId,parseCredentials(submissionContext),
          submissionContext.getCancelTokensWhenComplete()
          );
    }
  } catch (IOException ie) {
    LOG.warn(
        "Unable to add the application to the delegation token renewer.",
        ie);
    // Sending APP_REJECTED is fine, since we assume that the
    // RMApp is in NEW state and thus we havne't yet informed the
    // Scheduler about the existence of the application
    this.rmContext.getDispatcher().getEventHandler().handle(
        new RMAppRejectedEvent(applicationId, ie.getMessage()));
    throw RPCUtil.getRemoteException(ie);
  }

  // All done, start the RMApp
  this.rmContext.getDispatcher().getEventHandler().handle(
      new RMAppEvent(applicationId, isRecovered ? RMAppEventType.RECOVER:
          RMAppEventType.START));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; //导入依赖的package包/类
/**
 * Add a new application to the scheduler, with a given id, queue name, and
 * user. This will accept a new app even if the user or queue is above
 * configured limits, but the app will not be marked as runnable.
 */
protected synchronized void addApplication(ApplicationId applicationId,
    String queueName, String user) {
  if (queueName == null || queueName.isEmpty()) {
    String message = "Reject application " + applicationId +
            " submitted by user " + user + " with an empty queue name.";
    LOG.info(message);
    rmContext.getDispatcher().getEventHandler()
        .handle(new RMAppRejectedEvent(applicationId, message));
    return;
  }

  RMApp rmApp = rmContext.getRMApps().get(applicationId);
  FSLeafQueue queue = assignToQueue(rmApp, queueName, user);
  if (queue == null) {
    return;
  }

  // Enforce ACLs
  UserGroupInformation userUgi = UserGroupInformation.createRemoteUser(user);

  if (!queue.hasAccess(QueueACL.SUBMIT_APPLICATIONS, userUgi)
      && !queue.hasAccess(QueueACL.ADMINISTER_QUEUE, userUgi)) {
    String msg = "User " + userUgi.getUserName() +
            " cannot submit applications to queue " + queue.getName();
    LOG.info(msg);
    rmContext.getDispatcher().getEventHandler()
        .handle(new RMAppRejectedEvent(applicationId, msg));
    return;
  }

  SchedulerApplication application =
      new SchedulerApplication(queue, user);
  applications.put(applicationId, application);
  queue.getMetrics().submitApp(user);

  LOG.info("Accepted application " + applicationId + " from user: " + user
      + ", in queue: " + queueName + ", currently num of applications: "
      + applications.size());
  rmContext.getDispatcher().getEventHandler()
      .handle(new RMAppEvent(applicationId, RMAppEventType.APP_ACCEPTED));
}