@SuppressWarnings("unchecked")
  @Override
  public MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues(
      MoveApplicationAcrossQueuesRequest request) throws YarnException {
    ApplicationId applicationId = request.getApplicationId();

    UserGroupInformation callerUGI = getCallerUgi(applicationId,
        AuditConstants.MOVE_APP_REQUEST);
    RMApp application = verifyUserAccessForRMApp(applicationId, callerUGI,
        AuditConstants.MOVE_APP_REQUEST, ApplicationAccessType.MODIFY_APP,
        true);

    String targetQueue = request.getTargetQueue();
    if (!accessToTargetQueueAllowed(callerUGI, application, targetQueue)) {
      RMAuditLogger.logFailure(callerUGI.getShortUserName(),
          AuditConstants.MOVE_APP_REQUEST, "Target queue doesn't exist or user"
              + " doesn't have permissions to submit to target queue: "
              + targetQueue, "ClientRMService",
          AuditConstants.UNAUTHORIZED_USER, applicationId);
      throw RPCUtil.getRemoteException(new AccessControlException("User "
          + callerUGI.getShortUserName() + " cannot submit applications to"
          + " target queue or the target queue doesn't exist: "
          + targetQueue + " while moving " + applicationId));
    }

    // Moves only allowed when app is in a state that means it is tracked by
    // the scheduler. Introducing SUBMITTED state also to this list as there
    // could be a corner scenario that app may not be in Scheduler in SUBMITTED
    // state.
    if (!ACTIVE_APP_STATES.contains(application.getState())) {
      String msg = "App in " + application.getState() +
          " state cannot be moved.";
      RMAuditLogger.logFailure(callerUGI.getShortUserName(),
          AuditConstants.MOVE_APP_REQUEST, "UNKNOWN", "ClientRMService", msg);
      throw new YarnException(msg);
    }

    try {
      this.rmAppManager.moveApplicationAcrossQueue(
          application.getApplicationId(),
          request.getTargetQueue());
    } catch (YarnException ex) {
      RMAuditLogger.logFailure(callerUGI.getShortUserName(),
          AuditConstants.MOVE_APP_REQUEST, "UNKNOWN", "ClientRMService",
          ex.getMessage());
      throw ex;
    }

    RMAuditLogger.logSuccess(callerUGI.getShortUserName(), 
        AuditConstants.MOVE_APP_REQUEST, "ClientRMService" , applicationId);
    return recordFactory
        .newRecordInstance(MoveApplicationAcrossQueuesResponse.class);
  }
@Test
  public void testMoveApplicationAdminTargetQueue() throws Exception {
    ApplicationId applicationId = getApplicationId(1);
    UserGroupInformation aclUGI = UserGroupInformation.getCurrentUser();
    QueueACLsManager queueAclsManager = getQueueAclManager("allowed_queue",
        QueueACL.ADMINISTER_QUEUE, aclUGI);
    ApplicationACLsManager appAclsManager = getAppAclManager();
    ClientRMService rmService =
        createClientRMServiceForMoveApplicationRequest(applicationId,
            aclUGI.getShortUserName(), appAclsManager, queueAclsManager);

    // user is admin move to queue in acl
    MoveApplicationAcrossQueuesRequest moveAppRequest =
        MoveApplicationAcrossQueuesRequest.newInstance(applicationId,
            "allowed_queue");
    rmService.moveApplicationAcrossQueues(moveAppRequest);

    // user is admin move to queue not in acl
    moveAppRequest = MoveApplicationAcrossQueuesRequest.newInstance(
        applicationId, "not_allowed");

    try {
      rmService.moveApplicationAcrossQueues(moveAppRequest);
      Assert.fail("The request should fail with an AccessControlException");
    } catch (YarnException rex) {
      Assert.assertTrue("AccessControlException is expected",
          rex.getCause() instanceof AccessControlException);
    }

    // ACL is owned by "moveuser", move is performed as a different user
    aclUGI = UserGroupInformation.createUserForTesting("moveuser",
        new String[]{});
    queueAclsManager = getQueueAclManager("move_queue",
        QueueACL.ADMINISTER_QUEUE, aclUGI);
    appAclsManager = getAppAclManager();
    ClientRMService rmService2 =
        createClientRMServiceForMoveApplicationRequest(applicationId,
            aclUGI.getShortUserName(), appAclsManager, queueAclsManager);

    // no access to this queue
    MoveApplicationAcrossQueuesRequest moveAppRequest2 =
        MoveApplicationAcrossQueuesRequest.
            newInstance(applicationId, "move_queue");

    try {
      rmService2.moveApplicationAcrossQueues(moveAppRequest2);
      Assert.fail("The request should fail with an AccessControlException");
    } catch (YarnException rex) {
      Assert.assertTrue("AccessControlException is expected",
          rex.getCause() instanceof AccessControlException);
    }

    // execute the move as the acl owner
    // access to the queue OK: user allowed in this queue
    aclUGI.doAs(new PrivilegedExceptionAction<Object>() {
      @Override
      public Object run() throws Exception {
        return rmService2.moveApplicationAcrossQueues(moveAppRequest2);
      }
    });
  }