private <R> Map<SubClusterInfo, R> invokeConcurrent(Collection<SubClusterInfo> clusterIds,
      ClientMethod request, Class<R> clazz) throws YarnException {

    Map<SubClusterInfo, R> results = new HashMap<>();

    // If there is a sub-cluster access error,
    // we should choose whether to throw exception information according to user configuration.
    // Send the requests in parallel.
    CompletionService<SubClusterResult<R>> compSvc = new ExecutorCompletionService<>(threadpool);

    // This part of the code should be able to expose the accessed Exception information.
    // We use Pair to store related information. The left value of the Pair is the response,
    // and the right value is the exception.
    // If the request is normal, the response is not empty and the exception is empty;
    // if the request is abnormal, the response is empty and the exception is not empty.
    for (final SubClusterInfo info : clusterIds) {
      compSvc.submit(() -> {
        DefaultRequestInterceptorREST interceptor = getOrCreateInterceptorForSubCluster(
            info.getSubClusterId(), info.getRMWebServiceAddress());
        try {
          Method method = DefaultRequestInterceptorREST.class.
              getMethod(request.getMethodName(), request.getTypes());
          Object retObj = method.invoke(interceptor, request.getParams());
          R ret = clazz.cast(retObj);
          return new SubClusterResult<>(info, ret, null);
        } catch (Exception e) {
          LOG.error("SubCluster {} failed to call {} method.",
              info.getSubClusterId(), request.getMethodName(), e);
          return new SubClusterResult<>(info, null, e);
        }
      });
    }

    for (int i = 0; i < clusterIds.size(); i++) {
      SubClusterInfo subClusterInfo = null;
      try {
        Future<SubClusterResult<R>> future = compSvc.take();
        SubClusterResult<R> result = future.get();
        subClusterInfo = result.getSubClusterInfo();

        R response = result.getResponse();
        if (response != null) {
          results.put(subClusterInfo, response);
        }

        Exception exception = result.getException();
        if (exception != null) {
          throw exception;
        }
      } catch (Throwable e) {
        String subClusterId = subClusterInfo != null ?
            subClusterInfo.getSubClusterId().getId() : "UNKNOWN";
        LOG.error("SubCluster {} failed to {} report.", subClusterId, request.getMethodName(), e);
        // If allowPartialResult=false, it means that if an exception occurs in a subCluster,
        // an exception will be thrown directly.
        if (!allowPartialResult) {
          throw new YarnException("SubCluster " + subClusterId +
              " failed to " + request.getMethodName() + " report.", e);
        }
      }
    }

    return results;
  }
@Test
  public void testInvokeConcurrent() throws IOException, YarnException {

    // We design such a test case, we call the interceptor's getNodes interface,
    // this interface will generate the following test data
    // subCluster0 Node 0
    // subCluster1 Node 1
    // subCluster2 Node 2
    // subCluster3 Node 3
    // We use the returned data to verify whether the subClusterId
    // of the multi-thread call can match the node data
    Map<SubClusterInfo, NodesInfo> subClusterInfoNodesInfoMap =
        interceptor.invokeConcurrentGetNodeLabel();
    Assert.assertNotNull(subClusterInfoNodesInfoMap);
    Assert.assertEquals(4, subClusterInfoNodesInfoMap.size());

    subClusterInfoNodesInfoMap.forEach((subClusterInfo, nodesInfo) -> {
      String subClusterId = subClusterInfo.getSubClusterId().getId();
      List<NodeInfo> nodeInfos = nodesInfo.getNodes();
      Assert.assertNotNull(nodeInfos);
      Assert.assertEquals(1, nodeInfos.size());

      String expectNodeId = "Node " + subClusterId;
      String nodeId = nodeInfos.get(0).getNodeId();
      Assert.assertEquals(expectNodeId, nodeId);
    });
  }