@Override
  public Response replaceLabelsOnNodes(NodeToLabelsEntryList newNodeToLabels,
      HttpServletRequest hsr) throws IOException {

    // Step1. Check the parameters to ensure that the parameters are not empty.
    if (newNodeToLabels == null) {
      routerMetrics.incrReplaceLabelsOnNodesFailedRetrieved();
      throw new IllegalArgumentException("Parameter error, newNodeToLabels must not be empty.");
    }
    List<NodeToLabelsEntry> nodeToLabelsEntries = newNodeToLabels.getNodeToLabels();
    if (CollectionUtils.isEmpty(nodeToLabelsEntries)) {
      routerMetrics.incrReplaceLabelsOnNodesFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), REPLACE_LABELSONNODES, UNKNOWN,
          TARGET_WEB_SERVICE, "Parameter error, " +
          "nodeToLabelsEntries must not be empty.");
      throw new IllegalArgumentException("Parameter error, " +
         "nodeToLabelsEntries must not be empty.");
    }

    try {

      // Step2. We map the NodeId and NodeToLabelsEntry in the request.
      Map<String, NodeToLabelsEntry> nodeIdToLabels = new HashMap<>();
      newNodeToLabels.getNodeToLabels().forEach(nodeIdToLabel -> {
        String nodeId = nodeIdToLabel.getNodeId();
        nodeIdToLabels.put(nodeId, nodeIdToLabel);
      });

      // Step3. We map SubCluster with NodeToLabelsEntryList
      Map<SubClusterInfo, NodeToLabelsEntryList> subClusterToNodeToLabelsEntryList =
          new HashMap<>();
      nodeIdToLabels.forEach((nodeId, nodeToLabelsEntry) -> {
        SubClusterInfo subClusterInfo = getNodeSubcluster(nodeId);
        NodeToLabelsEntryList nodeToLabelsEntryList = subClusterToNodeToLabelsEntryList.
            getOrDefault(subClusterInfo, new NodeToLabelsEntryList());
        nodeToLabelsEntryList.getNodeToLabels().add(nodeToLabelsEntry);
        subClusterToNodeToLabelsEntryList.put(subClusterInfo, nodeToLabelsEntryList);
      });

      // Step4. Traverse the subCluster and call the replaceLabelsOnNodes interface.
      long startTime = clock.getTime();
      final HttpServletRequest hsrCopy = clone(hsr);
      StringBuilder builder = new StringBuilder();
      subClusterToNodeToLabelsEntryList.forEach((subClusterInfo, nodeToLabelsEntryList) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        try {
          DefaultRequestInterceptorREST interceptor = getOrCreateInterceptorForSubCluster(
              subClusterInfo);
          interceptor.replaceLabelsOnNodes(nodeToLabelsEntryList, hsrCopy);
          builder.append("subCluster-").append(subClusterId.getId()).append(":Success,");
        } catch (Exception e) {
          LOG.error("replaceLabelsOnNodes Failed. subClusterId = {}.", subClusterId, e);
          builder.append("subCluster-").append(subClusterId.getId()).append(":Failed,");
        }
      });
      long stopTime = clock.getTime();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), REPLACE_LABELSONNODES,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededReplaceLabelsOnNodesRetrieved(stopTime - startTime);

      // Step5. return call result.
      return Response.status(Status.OK).entity(builder.toString()).build();
    } catch (Exception e) {
      routerMetrics.incrReplaceLabelsOnNodesFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), REPLACE_LABELSONNODES, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      throw e;
    }
  }
@Test
  public void testReplaceLabelsOnNodes() throws Exception {
    // subCluster0 -> node0:0 -> label:NodeLabel0
    // subCluster1 -> node1:1 -> label:NodeLabel1
    // subCluster2 -> node2:2 -> label:NodeLabel2
    // subCluster3 -> node3:3 -> label:NodeLabel3
    NodeToLabelsEntryList nodeToLabelsEntryList = new NodeToLabelsEntryList();
    for (int i = 0; i < NUM_SUBCLUSTER; i++) {
      // labels
      List<String> labels = new ArrayList<>();
      labels.add("NodeLabel" + i);
      // nodes
      String nodeId = "node" + i + ":" + i;
      NodeToLabelsEntry nodeToLabelsEntry = new NodeToLabelsEntry(nodeId, labels);
      List<NodeToLabelsEntry> nodeToLabelsEntries = nodeToLabelsEntryList.getNodeToLabels();
      nodeToLabelsEntries.add(nodeToLabelsEntry);
    }

    // one of the results:
    // subCluster#0:Success;subCluster#1:Success;subCluster#3:Success;subCluster#2:Success;
    // We can't confirm the complete return order.
    Response response = interceptor.replaceLabelsOnNodes(nodeToLabelsEntryList, null);
    Assert.assertNotNull(response);
    Assert.assertEquals(200, response.getStatus());

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

    String entityValue = String.valueOf(entityObject);
    String[] entities = entityValue.split(",");
    Assert.assertNotNull(entities);
    Assert.assertEquals(4, entities.length);
    String expectValue =
        "subCluster-0:Success,subCluster-1:Success,subCluster-2:Success,subCluster-3:Success,";
    for (String entity : entities) {
      Assert.assertTrue(expectValue.contains(entity));
    }
  }