@Override
  public Response submitApplication(ApplicationSubmissionContextInfo newApp, HttpServletRequest hsr)
      throws AuthorizationException, IOException, InterruptedException {

    long startTime = clock.getTime();

    // We verify the parameters to ensure that newApp is not empty and
    // that the format of applicationId is correct.
    if (newApp == null || newApp.getApplicationId() == null) {
      routerMetrics.incrAppsFailedSubmitted();
      String errMsg = "Missing ApplicationSubmissionContextInfo or "
          + "applicationSubmissionContext information.";
      RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
          TARGET_WEB_SERVICE, errMsg);
      return Response.status(Status.BAD_REQUEST).entity(errMsg).build();
    }

    try {
      String applicationId = newApp.getApplicationId();
      RouterServerUtil.validateApplicationId(applicationId);
    } catch (IllegalArgumentException e) {
      routerMetrics.incrAppsFailedSubmitted();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
          TARGET_WEB_SERVICE, e.getMessage());
      return Response.status(Status.BAD_REQUEST).entity(e.getLocalizedMessage()).build();
    }

    List<SubClusterId> blackList = new ArrayList<>();
    try {
      int activeSubClustersCount = federationFacade.getActiveSubClustersCount();
      int actualRetryNums = Math.min(activeSubClustersCount, numSubmitRetries);
      Response response = ((FederationActionRetry<Response>) (retryCount) ->
          invokeSubmitApplication(newApp, blackList, hsr, retryCount)).
          runWithRetries(actualRetryNums, submitIntervalTime);
      if (response != null) {
        long stopTime = clock.getTime();
        routerMetrics.succeededAppsSubmitted(stopTime - startTime);
        return response;
      }
    } catch (Exception e) {
      routerMetrics.incrAppsFailedSubmitted();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
          TARGET_WEB_SERVICE, e.getMessage());
      return Response.status(Status.SERVICE_UNAVAILABLE).entity(e.getLocalizedMessage()).build();
    }

    routerMetrics.incrAppsFailedSubmitted();
    String errMsg = String.format("Application %s with appId %s failed to be submitted.",
        newApp.getApplicationName(), newApp.getApplicationId());
    LOG.error(errMsg);
    RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_NEW_APP, UNKNOWN,
        TARGET_WEB_SERVICE, errMsg);
    return Response.status(Status.SERVICE_UNAVAILABLE).entity(errMsg).build();
  }
@Test
  public void testSubmitApplication()
      throws YarnException, IOException, InterruptedException {

    ApplicationId appId =
        ApplicationId.newInstance(Time.now(), 1);

    ApplicationSubmissionContextInfo context =
        new ApplicationSubmissionContextInfo();
    context.setApplicationId(appId.toString());

    Response response = interceptor.submitApplication(context, null);
    Assert.assertEquals(ACCEPTED, response.getStatus());
    SubClusterId ci = (SubClusterId) response.getEntity();

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