@Override
  public Response addToClusterNodeLabels(NodeLabelsInfo newNodeLabels,
      HttpServletRequest hsr) throws Exception {

    if (newNodeLabels == null) {
      routerMetrics.incrAddToClusterNodeLabelsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), ADD_TO_CLUSTER_NODELABELS, UNKNOWN,
          TARGET_WEB_SERVICE, "Parameter error, the newNodeLabels is null.");
      throw new IllegalArgumentException("Parameter error, the newNodeLabels is null.");
    }

    List<NodeLabelInfo> nodeLabelInfos = newNodeLabels.getNodeLabelsInfo();
    if (CollectionUtils.isEmpty(nodeLabelInfos)) {
      routerMetrics.incrAddToClusterNodeLabelsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), ADD_TO_CLUSTER_NODELABELS, UNKNOWN,
          TARGET_WEB_SERVICE, "Parameter error, the nodeLabelsInfo is null or empty.");
      throw new IllegalArgumentException("Parameter error, the nodeLabelsInfo is null or empty.");
    }

    try {
      long startTime = clock.getTime();
      Collection<SubClusterInfo> subClustersActives = federationFacade.getActiveSubClusters();
      final HttpServletRequest hsrCopy = clone(hsr);
      Class[] argsClasses = new Class[]{NodeLabelsInfo.class, HttpServletRequest.class};
      Object[] args = new Object[]{newNodeLabels, hsrCopy};
      ClientMethod remoteMethod = new ClientMethod("addToClusterNodeLabels", argsClasses, args);
      Map<SubClusterInfo, Response> responseInfoMap =
          invokeConcurrent(subClustersActives, remoteMethod, Response.class);
      StringBuilder buffer = new StringBuilder();
      // SubCluster-0:SUCCESS,SubCluster-1:SUCCESS
      responseInfoMap.forEach((subClusterInfo, response) ->
          buildAppendMsg(subClusterInfo, buffer, response));
      long stopTime = clock.getTime();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), ADD_TO_CLUSTER_NODELABELS,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededAddToClusterNodeLabelsRetrieved((stopTime - startTime));
      return Response.status(Status.OK).entity(buffer.toString()).build();
    } catch (NotFoundException e) {
      routerMetrics.incrAddToClusterNodeLabelsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), ADD_TO_CLUSTER_NODELABELS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowIOException("get all active sub cluster(s) error.", e);
    } catch (YarnException e) {
      routerMetrics.incrAddToClusterNodeLabelsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), ADD_TO_CLUSTER_NODELABELS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowIOException("addToClusterNodeLabels with yarn error.", e);
    }

    routerMetrics.incrAddToClusterNodeLabelsFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), ADD_TO_CLUSTER_NODELABELS, UNKNOWN,
        TARGET_WEB_SERVICE, "addToClusterNodeLabels Failed.");
    throw new RuntimeException("addToClusterNodeLabels Failed.");
  }
@Test
  public void testAddToClusterNodeLabels1() throws Exception {
    // In this test, we try to add ALL label, all subClusters will return success.
    NodeLabelsInfo nodeLabelsInfo = new NodeLabelsInfo();
    NodeLabelInfo nodeLabelInfo = new NodeLabelInfo("ALL", true);
    nodeLabelsInfo.getNodeLabelsInfo().add(nodeLabelInfo);

    Response response = interceptor.addToClusterNodeLabels(nodeLabelsInfo, null);
    Assert.assertNotNull(response);

    Object entityObj = response.getEntity();
    Assert.assertNotNull(entityObj);

    String entity = String.valueOf(entityObj);
    String[] entities = StringUtils.split(entity, ",");
    Assert.assertNotNull(entities);
    Assert.assertEquals(4, entities.length);

    // The order in which the cluster returns messages is uncertain,
    // we confirm the result by contains
    String expectedMsg =
        "SubCluster-0:SUCCESS,SubCluster-1:SUCCESS,SubCluster-2:SUCCESS,SubCluster-3:SUCCESS";
    Arrays.stream(entities).forEach(item -> Assert.assertTrue(expectedMsg.contains(item)));
  }