@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 testUpdateSchedulerConfiguration()
      throws AuthorizationException, InterruptedException {
    SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
    updateInfo.setSubClusterId("1");
    Map<String, String> goodUpdateMap = new HashMap<>();
    goodUpdateMap.put("goodKey", "goodVal");
    QueueConfigInfo goodUpdateInfo = new
        QueueConfigInfo("root.default", goodUpdateMap);
    updateInfo.getUpdateQueueInfo().add(goodUpdateInfo);
    Response response = interceptor.updateSchedulerConfiguration(updateInfo, null);

    Assert.assertNotNull(response);
    Assert.assertEquals(OK, response.getStatus());

    String expectMsg = "Configuration change successfully applied.";
    Object entity = response.getEntity();
    Assert.assertNotNull(entity);

    String entityMsg = String.valueOf(entity);
    Assert.assertEquals(expectMsg, entityMsg);
  }