@Override
  public String dumpSchedulerLogs(String time, HttpServletRequest hsr)
      throws IOException {

    // Step1. We will check the time parameter to
    // ensure that the time parameter is not empty and greater than 0.

    if (StringUtils.isBlank(time)) {
      routerMetrics.incrDumpSchedulerLogsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), DUMP_SCHEDULERLOGS, UNKNOWN,
          TARGET_WEB_SERVICE, "Parameter error, the time is empty or null.");
      throw new IllegalArgumentException("Parameter error, the time is empty or null.");
    }

    try {
      int period = Integer.parseInt(time);
      if (period <= 0) {
        throw new IllegalArgumentException("time must be greater than 0.");
      }
    } catch (NumberFormatException e) {
      routerMetrics.incrDumpSchedulerLogsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), DUMP_SCHEDULERLOGS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      throw new IllegalArgumentException("time must be a number.");
    } catch (IllegalArgumentException e) {
      routerMetrics.incrDumpSchedulerLogsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), DUMP_SCHEDULERLOGS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      throw e;
    }

    // Step2. Call dumpSchedulerLogs of each subcluster.
    try {
      long startTime = clock.getTime();
      Collection<SubClusterInfo> subClustersActive = federationFacade.getActiveSubClusters();
      final HttpServletRequest hsrCopy = clone(hsr);
      Class[] argsClasses = new Class[]{String.class, HttpServletRequest.class};
      Object[] args = new Object[]{time, hsrCopy};
      ClientMethod remoteMethod = new ClientMethod("dumpSchedulerLogs", argsClasses, args);
      Map<SubClusterInfo, String> dumpSchedulerLogsMap = invokeConcurrent(
          subClustersActive, remoteMethod, String.class);
      StringBuilder stringBuilder = new StringBuilder();
      dumpSchedulerLogsMap.forEach((subClusterInfo, msg) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        stringBuilder.append("subClusterId")
            .append(subClusterId).append(" : ").append(msg).append("; ");
      });
      long stopTime = clock.getTime();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), DUMP_SCHEDULERLOGS,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededDumpSchedulerLogsRetrieved(stopTime - startTime);
      return stringBuilder.toString();
    } catch (IllegalArgumentException e) {
      routerMetrics.incrDumpSchedulerLogsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), DUMP_SCHEDULERLOGS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException(e,
          "Unable to dump SchedulerLogs by time: %s.", time);
    } catch (YarnException e) {
      routerMetrics.incrDumpSchedulerLogsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), DUMP_SCHEDULERLOGS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException(e,
          "dumpSchedulerLogs by time = %s error .", time);
    }

    routerMetrics.incrDumpSchedulerLogsFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), DUMP_SCHEDULERLOGS, UNKNOWN,
        TARGET_WEB_SERVICE, "dumpSchedulerLogs Failed.");
    throw new RuntimeException("dumpSchedulerLogs Failed.");
  }
@Test
  public void testDumpSchedulerLogs() throws Exception {
    HttpServletRequest mockHsr = mockHttpServletRequestByUserName("admin");
    String dumpSchedulerLogsMsg = interceptor.dumpSchedulerLogs("1", mockHsr);

    // We cannot guarantee the calling order of the sub-clusters,
    // We guarantee that the returned result contains the information of each subCluster.
    Assert.assertNotNull(dumpSchedulerLogsMsg);
    subClusters.forEach(subClusterId -> {
      String subClusterMsg =
          "subClusterId" + subClusterId + " : Capacity scheduler logs are being created.; ";
      Assert.assertTrue(dumpSchedulerLogsMsg.contains(subClusterMsg));
    });
  }