@Override
  public MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues(
      MoveApplicationAcrossQueuesRequest request)
      throws YarnException, IOException {
    if (request == null || request.getApplicationId() == null || request.getTargetQueue() == null) {
      routerMetrics.incrMoveApplicationAcrossQueuesFailedRetrieved();
      String msg = "Missing moveApplicationAcrossQueues request or " +
          "applicationId or target queue.";
      RouterAuditLogger.logFailure(user.getShortUserName(), MOVE_APPLICATION_ACROSS_QUEUES, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, msg);
      RouterServerUtil.logAndThrowException(msg);
    }

    long startTime = clock.getTime();
    SubClusterId subClusterId = null;

    ApplicationId applicationId = request.getApplicationId();
    try {
      subClusterId = federationFacade
          .getApplicationHomeSubCluster(applicationId);
    } catch (YarnException e) {
      routerMetrics.incrMoveApplicationAcrossQueuesFailedRetrieved();
      String errMsgFormat = "Application %s does not exist in FederationStateStore.";
      RouterAuditLogger.logFailure(user.getShortUserName(), MOVE_APPLICATION_ACROSS_QUEUES, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, String.format(errMsgFormat, applicationId));
      RouterServerUtil.logAndThrowException(e, errMsgFormat, applicationId);
    }

    ApplicationClientProtocol clientRMProxy = getClientRMProxyForSubCluster(subClusterId);
    MoveApplicationAcrossQueuesResponse response = null;
    try {
      response = clientRMProxy.moveApplicationAcrossQueues(request);
    } catch (Exception e) {
      routerMetrics.incrMoveApplicationAcrossQueuesFailedRetrieved();
      RouterServerUtil.logAndThrowException(e,
          "Unable to moveApplicationAcrossQueues for %s to SubCluster %s.", applicationId,
          subClusterId.getId());
    }

    if (response == null) {
      LOG.error("No response when moveApplicationAcrossQueues "
           + "the applicationId {} to Queue {} In SubCluster {}.",
           request.getApplicationId(), request.getTargetQueue(), subClusterId.getId());
    }

    long stopTime = clock.getTime();
    RouterAuditLogger.logSuccess(user.getShortUserName(), MOVE_APPLICATION_ACROSS_QUEUES,
        TARGET_CLIENT_RM_SERVICE, applicationId, subClusterId);
    routerMetrics.succeededMoveApplicationAcrossQueuesRetrieved(stopTime - startTime);
    return response;
  }
@Test
  public void testMoveApplicationAcrossQueues() throws Exception {
    LOG.info("Test FederationClientInterceptor : MoveApplication AcrossQueues request.");

    // null request
    LambdaTestUtils.intercept(YarnException.class, "Missing moveApplicationAcrossQueues request " +
        "or applicationId or target queue.", () -> interceptor.moveApplicationAcrossQueues(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);
    MockNM nm = interceptor.getMockNMs().get(subClusterId);
    nm.nodeHeartbeat(true);
    MockRM.waitForState(rmApp.getCurrentAppAttempt(), RMAppAttemptState.ALLOCATED);
    mockRM.sendAMLaunched(rmApp.getCurrentAppAttempt().getAppAttemptId());

    MoveApplicationAcrossQueuesRequest acrossQueuesRequest =
        MoveApplicationAcrossQueuesRequest.newInstance(appId, "root.target");
    MoveApplicationAcrossQueuesResponse acrossQueuesResponse =
        interceptor.moveApplicationAcrossQueues(acrossQueuesRequest);

    Assert.assertNotNull(acrossQueuesResponse);
  }