@Override
  public SubmitApplicationResponse submitApplication(
      SubmitApplicationRequest request) throws YarnException, IOException {

    if (request == null || request.getApplicationSubmissionContext() == null
        || request.getApplicationSubmissionContext().getApplicationId() == null) {
      routerMetrics.incrAppsFailedSubmitted();
      String errMsg =
          "Missing submitApplication request or applicationSubmissionContext information.";
      RouterAuditLogger.logFailure(user.getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, errMsg);
      RouterServerUtil.logAndThrowException(errMsg, null);
    }

    long startTime = clock.getTime();
    ApplicationId applicationId =
        request.getApplicationSubmissionContext().getApplicationId();
    List<SubClusterId> blacklist = new ArrayList<>();

    try {

      // We need to handle this situation,
      // the user will provide us with an expected submitRetries,
      // but if the number of Active SubClusters is less than this number at this time,
      // we should provide a high number of retry according to the number of Active SubClusters.
      int activeSubClustersCount = federationFacade.getActiveSubClustersCount();
      int actualRetryNums = Math.min(activeSubClustersCount, numSubmitRetries);

      // Try calling the SubmitApplication method
      SubmitApplicationResponse response =
          ((FederationActionRetry<SubmitApplicationResponse>) (retryCount) ->
          invokeSubmitApplication(blacklist, request, retryCount)).
          runWithRetries(actualRetryNums, submitIntervalTime);

      if (response != null) {
        long stopTime = clock.getTime();
        routerMetrics.succeededAppsSubmitted(stopTime - startTime);
        return response;
      }

    } catch (Exception e) {
      routerMetrics.incrAppsFailedSubmitted();
      RouterAuditLogger.logFailure(user.getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, e.getMessage(), applicationId);
      RouterServerUtil.logAndThrowException(e.getMessage(), e);
    }

    routerMetrics.incrAppsFailedSubmitted();
    String msg = String.format("Application %s with appId %s failed to be submitted.",
        request.getApplicationSubmissionContext().getApplicationName(), applicationId);
    RouterAuditLogger.logFailure(user.getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
        TARGET_CLIENT_RM_SERVICE, msg, applicationId);
    throw new YarnException(msg);
  }
@Test
  public void testSubmitApplication()
      throws YarnException, IOException {
    LOG.info("Test FederationClientInterceptor: Submit Application.");

    ApplicationId appId = ApplicationId.newInstance(System.currentTimeMillis(), 1);
    SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);

    SubmitApplicationResponse response = interceptor.submitApplication(request);

    Assert.assertNotNull(response);
    SubClusterId scIdResult = stateStoreUtil.queryApplicationHomeSC(appId);
    Assert.assertNotNull(scIdResult);
    Assert.assertTrue(subClusters.contains(scIdResult));
  }