@Override
  public ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request) throws YarnException, IOException {

    if (request == null || request.getReservationId() == null
        || request.getReservationDefinition() == null || request.getQueue() == null) {
      routerMetrics.incrSubmitReservationFailedRetrieved();
      String msg = "Missing submitReservation request or reservationId " +
          "or reservation definition or queue.";
      RouterAuditLogger.logFailure(user.getShortUserName(), SUBMIT_RESERVATION, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, msg);
      RouterServerUtil.logAndThrowException(msg, null);
    }

    long startTime = clock.getTime();
    ReservationId reservationId = request.getReservationId();

    for (int i = 0; i < numSubmitRetries; i++) {
      try {
        // First, Get SubClusterId according to specific strategy.
        SubClusterId subClusterId = policyFacade.getReservationHomeSubCluster(request);
        LOG.info("submitReservation ReservationId {} try #{} on SubCluster {}.",
            reservationId, i, subClusterId);
        ReservationHomeSubCluster reservationHomeSubCluster =
            ReservationHomeSubCluster.newInstance(reservationId, subClusterId);

        // Second, determine whether the current ReservationId has a corresponding subCluster.
        // If it does not exist, add it. If it exists, update it.
        Boolean exists = existsReservationHomeSubCluster(reservationId);

        // We may encounter the situation of repeated submission of Reservation,
        // at this time we should try to use the reservation that has been allocated
        // !exists indicates that the reservation does not exist and needs to be added
        // i==0, mainly to consider repeated submissions,
        // so the first time to apply for reservation, try to use the original reservation
        if (!exists || i == 0) {
          addReservationHomeSubCluster(reservationId, reservationHomeSubCluster);
        } else {
          updateReservationHomeSubCluster(subClusterId, reservationId, reservationHomeSubCluster);
        }

        // Third, Submit a Reservation request to the subCluster
        ApplicationClientProtocol clientRMProxy = getClientRMProxyForSubCluster(subClusterId);
        ReservationSubmissionResponse response = clientRMProxy.submitReservation(request);
        if (response != null) {
          LOG.info("Reservation {} submitted on subCluster {}.", reservationId, subClusterId);
          long stopTime = clock.getTime();
          routerMetrics.succeededSubmitReservationRetrieved(stopTime - startTime);
          RouterAuditLogger.logSuccess(user.getShortUserName(), SUBMIT_RESERVATION,
              TARGET_CLIENT_RM_SERVICE);
          return response;
        }
      } catch (Exception e) {
        LOG.warn("Unable to submit(try #{}) the Reservation {}.", i, reservationId, e);
      }
    }

    routerMetrics.incrSubmitReservationFailedRetrieved();
    String msg = String.format("Reservation %s failed to be submitted.", reservationId);
    RouterAuditLogger.logFailure(user.getShortUserName(), SUBMIT_RESERVATION, UNKNOWN,
        TARGET_CLIENT_RM_SERVICE, msg);
    throw new YarnException(msg);
  }
@Test
  public void testSubmitReservationEmptyRequest() throws Exception {
    LOG.info("Test FederationClientInterceptor : SubmitReservation request empty.");

    String errorMsg =
        "Missing submitReservation request or reservationId or reservation definition or queue.";

    // null request1
    LambdaTestUtils.intercept(YarnException.class, errorMsg,
        () -> interceptor.submitReservation(null));

    // null request2
    ReservationSubmissionRequest request2 =
        ReservationSubmissionRequest.newInstance(null, null, null);
    LambdaTestUtils.intercept(YarnException.class, errorMsg,
        () -> interceptor.submitReservation(request2));

    // null request3
    ReservationSubmissionRequest request3 =
        ReservationSubmissionRequest.newInstance(null, "q1", null);
    LambdaTestUtils.intercept(YarnException.class, errorMsg,
        () -> interceptor.submitReservation(request3));

    // null request4
    ReservationId reservationId = ReservationId.newInstance(Time.now(), 1);
    ReservationSubmissionRequest request4 =
        ReservationSubmissionRequest.newInstance(null, null,  reservationId);
    LambdaTestUtils.intercept(YarnException.class, errorMsg,
        () -> interceptor.submitReservation(request4));

    // null request5
    long arrival = Time.now();
    long deadline = arrival + (int)(DEFAULT_DURATION * 1.1);

    ReservationRequest rRequest = ReservationRequest.newInstance(
        Resource.newInstance(1024, 1), 1, 1, DEFAULT_DURATION);
    ReservationRequest[] rRequests = new ReservationRequest[] {rRequest};
    ReservationDefinition rDefinition = createReservationDefinition(arrival, deadline, rRequests,
        ReservationRequestInterpreter.R_ALL, "u1");
    ReservationSubmissionRequest request5 =
        ReservationSubmissionRequest.newInstance(rDefinition, null,  reservationId);
    LambdaTestUtils.intercept(YarnException.class, errorMsg,
        () -> interceptor.submitReservation(request5));
  }