@Override
  public GetNewReservationResponse getNewReservation(
      GetNewReservationRequest request) throws YarnException, IOException {

    if (request == null) {
      routerMetrics.incrGetNewReservationFailedRetrieved();
      String errMsg = "Missing getNewReservation request.";
      RouterAuditLogger.logFailure(user.getShortUserName(), GET_NEW_RESERVATION, UNKNOWN,
          TARGET_CLIENT_RM_SERVICE, errMsg);
      RouterServerUtil.logAndThrowException(errMsg, null);
    }

    long startTime = clock.getTime();
    Map<SubClusterId, SubClusterInfo> subClustersActive = federationFacade.getSubClusters(true);

    for (int i = 0; i < numSubmitRetries; ++i) {
      SubClusterId subClusterId = getRandomActiveSubCluster(subClustersActive);
      LOG.info("getNewReservation try #{} on SubCluster {}.", i, subClusterId);
      ApplicationClientProtocol clientRMProxy = getClientRMProxyForSubCluster(subClusterId);
      try {
        GetNewReservationResponse response = clientRMProxy.getNewReservation(request);
        if (response != null) {
          long stopTime = clock.getTime();
          routerMetrics.succeededGetNewReservationRetrieved(stopTime - startTime);
          RouterAuditLogger.logSuccess(user.getShortUserName(), GET_NEW_RESERVATION,
              TARGET_CLIENT_RM_SERVICE);
          return response;
        }
      } catch (Exception e) {
        String logFormatted = "Unable to create a new Reservation in SubCluster {}.";
        LOG.warn(logFormatted, subClusterId.getId(), e);
        RouterAuditLogger.logFailure(user.getShortUserName(), GET_NEW_RESERVATION, UNKNOWN,
            TARGET_CLIENT_RM_SERVICE, logFormatted, subClusterId.getId());
        subClustersActive.remove(subClusterId);
      }
    }

    routerMetrics.incrGetNewReservationFailedRetrieved();
    String errMsg = "Failed to create a new reservation.";
    RouterAuditLogger.logFailure(user.getShortUserName(), GET_NEW_RESERVATION, UNKNOWN,
        TARGET_CLIENT_RM_SERVICE, errMsg);
    throw new YarnException(errMsg);
  }
@Test
  public void testGetNewReservation() throws Exception {
    LOG.info("Test FederationClientInterceptor : Get NewReservation request.");

    // null request
    LambdaTestUtils.intercept(YarnException.class,
        "Missing getNewReservation request.", () -> interceptor.getNewReservation(null));

    // normal request
    GetNewReservationRequest request = GetNewReservationRequest.newInstance();
    GetNewReservationResponse response = interceptor.getNewReservation(request);
    Assert.assertNotNull(response);

    ReservationId reservationId = response.getReservationId();
    Assert.assertNotNull(reservationId);
    Assert.assertTrue(reservationId.toString().contains("reservation"));
    Assert.assertEquals(reservationId.getClusterTimestamp(), ResourceManager.getClusterTimeStamp());
  }