@Override
  public Response submitReservation(ReservationSubmissionRequestInfo resContext,
      HttpServletRequest hsr) throws AuthorizationException, IOException, InterruptedException {
    long startTime = clock.getTime();
    if (resContext == null || resContext.getReservationId() == null
        || resContext.getReservationDefinition() == null || resContext.getQueue() == null) {
      routerMetrics.incrSubmitReservationFailedRetrieved();
      String errMsg = "Missing submitReservation resContext or reservationId " +
          "or reservation definition or queue.";
      RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_RESERVATION,
          UNKNOWN, TARGET_WEB_SERVICE, errMsg);
      return Response.status(Status.BAD_REQUEST).entity(errMsg).build();
    }

    // Check that the resId format is accurate
    String resId = resContext.getReservationId();
    try {
      RouterServerUtil.validateReservationId(resId);
    } catch (IllegalArgumentException e) {
      routerMetrics.incrSubmitReservationFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_RESERVATION,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      throw e;
    }

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

    routerMetrics.incrSubmitReservationFailedRetrieved();
    String msg = String.format("Reservation %s failed to be submitted.", resId);
    RouterAuditLogger.logFailure(getUser().getShortUserName(), SUBMIT_RESERVATION,
        UNKNOWN, TARGET_WEB_SERVICE, msg);
    return Response.status(Status.SERVICE_UNAVAILABLE).entity(msg).build();
  }
@Test
  public void testSubmitReservation() throws Exception {

    // submit reservation
    ReservationId reservationId = ReservationId.newInstance(Time.now(), 2);
    Response response = submitReservation(reservationId);
    Assert.assertNotNull(response);
    Assert.assertEquals(Status.ACCEPTED.getStatusCode(), response.getStatus());

    String applyReservationId = reservationId.toString();
    Response reservationResponse = interceptor.listReservation(
        QUEUE_DEDICATED_FULL, applyReservationId, -1, -1, false, null);
    Assert.assertNotNull(reservationResponse);

    Object entity = reservationResponse.getEntity();
    Assert.assertNotNull(entity);
    Assert.assertNotNull(entity instanceof ReservationListInfo);

    Assert.assertTrue(entity instanceof ReservationListInfo);
    ReservationListInfo listInfo = (ReservationListInfo) entity;
    Assert.assertNotNull(listInfo);

    List<ReservationInfo> reservationInfos = listInfo.getReservations();
    Assert.assertNotNull(reservationInfos);
    Assert.assertEquals(1, reservationInfos.size());

    ReservationInfo reservationInfo = reservationInfos.get(0);
    Assert.assertNotNull(reservationInfo);
    Assert.assertEquals(reservationInfo.getReservationId(), applyReservationId);
  }