@Override
  public Response updateSchedulerConfiguration(SchedConfUpdateInfo mutationInfo,
      HttpServletRequest hsr) throws AuthorizationException, InterruptedException {

    // Make Sure mutationInfo is not null.
    if (mutationInfo == null) {
      routerMetrics.incrUpdateSchedulerConfigurationFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), UPDATE_SCHEDULER_CONFIGURATION,
          UNKNOWN, TARGET_WEB_SERVICE,
          "Parameter error, the schedConfUpdateInfo is empty or null.");
      throw new IllegalArgumentException(
          "Parameter error, the schedConfUpdateInfo is empty or null.");
    }

    // In federated mode, we may have a mix of multiple schedulers.
    // In order to ensure accurate update scheduler configuration,
    // we need users to explicitly set subClusterId.
    String pSubClusterId = mutationInfo.getSubClusterId();
    if (StringUtils.isBlank(pSubClusterId)) {
      routerMetrics.incrUpdateSchedulerConfigurationFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), UPDATE_SCHEDULER_CONFIGURATION,
          UNKNOWN, TARGET_WEB_SERVICE,
          "Parameter error, the subClusterId is empty or null.");
      throw new IllegalArgumentException("Parameter error, " +
          "the subClusterId is empty or null.");
    }

    // Get the subClusterInfo , then update the scheduler configuration.
    try {
      long startTime = clock.getTime();
      SubClusterInfo subClusterInfo = getActiveSubCluster(pSubClusterId);
      DefaultRequestInterceptorREST interceptor = getOrCreateInterceptorForSubCluster(
          subClusterInfo.getSubClusterId(), subClusterInfo.getRMWebServiceAddress());
      Response response = interceptor.updateSchedulerConfiguration(mutationInfo, hsr);
      if (response != null) {
        long endTime = clock.getTime();
        routerMetrics.succeededUpdateSchedulerConfigurationRetrieved(endTime - startTime);
        RouterAuditLogger.logSuccess(getUser().getShortUserName(), UPDATE_SCHEDULER_CONFIGURATION,
            TARGET_WEB_SERVICE);
        return Response.status(response.getStatus()).entity(response.getEntity()).build();
      }
    } catch (NotFoundException e) {
      routerMetrics.incrUpdateSchedulerConfigurationFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), UPDATE_SCHEDULER_CONFIGURATION,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException(e,
          "Get subCluster error. subClusterId = %s", pSubClusterId);
    } catch (Exception e) {
      routerMetrics.incrUpdateSchedulerConfigurationFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), UPDATE_SCHEDULER_CONFIGURATION,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException(e,
          "UpdateSchedulerConfiguration error. subClusterId = %s", pSubClusterId);
    }

    routerMetrics.incrUpdateSchedulerConfigurationFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), UPDATE_SCHEDULER_CONFIGURATION,
        UNKNOWN, TARGET_WEB_SERVICE, "UpdateSchedulerConfiguration Failed.");
    throw new RuntimeException("UpdateSchedulerConfiguration error. subClusterId = "
        + pSubClusterId);
  }
@Test
  public void testUpdateSchedulerConfigurationErrorMsg() throws Exception {
    SchedConfUpdateInfo mutationInfo = new SchedConfUpdateInfo();
    LambdaTestUtils.intercept(IllegalArgumentException.class,
        "Parameter error, the subClusterId is empty or null.",
        () -> interceptor.updateSchedulerConfiguration(mutationInfo, null));

    LambdaTestUtils.intercept(IllegalArgumentException.class,
        "Parameter error, the schedConfUpdateInfo is empty or null.",
        () -> interceptor.updateSchedulerConfiguration(null, null));
  }