@Override
  public FailApplicationAttemptResponse failApplicationAttempt(
      FailApplicationAttemptRequest request) throws YarnException, IOException {
    if (request == null || request.getApplicationAttemptId() == null
          || request.getApplicationAttemptId().getApplicationId() == null) {
      routerMetrics.incrFailAppAttemptFailedRetrieved();
      String msg = "Missing failApplicationAttempt request or applicationId " +
          "or applicationAttemptId information.";
      RouterAuditLogger.logFailure(user.getShortUserName(), FAIL_APPLICATIONATTEMPT, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, msg);
      RouterServerUtil.logAndThrowException(msg, null);
    }
    long startTime = clock.getTime();
    SubClusterId subClusterId = null;
    ApplicationAttemptId applicationAttemptId = request.getApplicationAttemptId();
    ApplicationId applicationId = applicationAttemptId.getApplicationId();

    try {
      subClusterId = getApplicationHomeSubCluster(applicationId);
    } catch (YarnException e) {
      routerMetrics.incrFailAppAttemptFailedRetrieved();
      String msg = "ApplicationAttempt " +
          applicationAttemptId + " belongs to Application " + applicationId +
          " does not exist in FederationStateStore.";
      RouterAuditLogger.logFailure(user.getShortUserName(), FAIL_APPLICATIONATTEMPT, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, msg);
      RouterServerUtil.logAndThrowException(msg, e);
    }

    ApplicationClientProtocol clientRMProxy = getClientRMProxyForSubCluster(subClusterId);
    FailApplicationAttemptResponse response = null;
    try {
      response = clientRMProxy.failApplicationAttempt(request);
    } catch (Exception e) {
      routerMetrics.incrFailAppAttemptFailedRetrieved();
      String msg = "Unable to get the applicationAttempt report for " +
          applicationAttemptId + " to SubCluster " + subClusterId;
      RouterAuditLogger.logFailure(user.getShortUserName(), FAIL_APPLICATIONATTEMPT, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, msg);
      RouterServerUtil.logAndThrowException(msg, e);
    }

    if (response == null) {
      LOG.error("No response when attempting to retrieve the report of " +
          "the applicationAttempt {} to SubCluster {}.",
          request.getApplicationAttemptId(), subClusterId.getId());
    }

    long stopTime = clock.getTime();
    routerMetrics.succeededFailAppAttemptRetrieved(stopTime - startTime);
    RouterAuditLogger.logSuccess(user.getShortUserName(), FAIL_APPLICATIONATTEMPT,
        TARGET_CLIENT_RM_SERVICE, applicationId, subClusterId);
    return response;
  }
@Test
  public void testFailApplicationAttempt() throws Exception {
    LOG.info("Test FederationClientInterceptor : Fail Application Attempt request.");

    // null request
    LambdaTestUtils.intercept(YarnException.class, "Missing failApplicationAttempt request " +
        "or applicationId or applicationAttemptId information.",
        () -> interceptor.failApplicationAttempt(null));

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

    // Submit the application
    SubmitApplicationResponse response = interceptor.submitApplication(request);

    Assert.assertNotNull(response);
    Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));

    SubClusterId subClusterId = interceptor.getApplicationHomeSubCluster(appId);
    Assert.assertNotNull(subClusterId);

    MockRM mockRM = interceptor.getMockRMs().get(subClusterId);
    mockRM.waitForState(appId, RMAppState.ACCEPTED);
    RMApp rmApp = mockRM.getRMContext().getRMApps().get(appId);
    mockRM.waitForState(rmApp.getCurrentAppAttempt().getAppAttemptId(),
        RMAppAttemptState.SCHEDULED);

    // Call GetApplicationAttempts
    GetApplicationAttemptsRequest attemptsRequest =
        GetApplicationAttemptsRequest.newInstance(appId);
    GetApplicationAttemptsResponse attemptsResponse =
        interceptor.getApplicationAttempts(attemptsRequest);
    Assert.assertNotNull(attemptsResponse);

    ApplicationAttemptId attemptId = attemptsResponse.getApplicationAttemptList().
        get(0).getApplicationAttemptId();

    FailApplicationAttemptRequest requestFailAppAttempt =
        FailApplicationAttemptRequest.newInstance(attemptId);
    FailApplicationAttemptResponse responseFailAppAttempt =
        interceptor.failApplicationAttempt(requestFailAppAttempt);

    Assert.assertNotNull(responseFailAppAttempt);
  }