class FederationInterceptorREST extends AbstractRESTRequestInterceptor {

  @Override
  public NodesInfo getNodes(String states) {

    NodesInfo nodes = new NodesInfo();

    Map<SubClusterId, SubClusterInfo> subClustersActive = null;
    try {
      subClustersActive = federationFacade.getSubClusters(true);
    } catch (YarnException e) {
      LOG.error("Cannot get nodes: {}", e.getMessage());
      return new NodesInfo();
    }

    // Send the requests in parallel
    CompletionService<NodesInfo> compSvc =
        new ExecutorCompletionService<NodesInfo>(this.threadpool);

    for (final SubClusterInfo info : subClustersActive.values()) {
      compSvc.submit(new Callable<NodesInfo>() {
        @Override
        public NodesInfo call() {
          DefaultRequestInterceptorREST interceptor =
              getOrCreateInterceptorForSubCluster(
                  info.getSubClusterId(), info.getRMWebServiceAddress());
          try {
            NodesInfo nodesInfo = interceptor.getNodes(states);
            return nodesInfo;
          } catch (Exception e) {
            LOG.error("Subcluster {} failed to return nodesInfo.",
                info.getSubClusterId());
            return null;
          }
        }
      });
    }

    // Collect all the responses in parallel

    for (int i = 0; i < subClustersActive.size(); i++) {
      try {
        Future<NodesInfo> future = compSvc.take();
        NodesInfo nodesResponse = future.get();

        if (nodesResponse != null) {
          nodes.addAll(nodesResponse.getNodes());
        }
      } catch (Throwable e) {
        LOG.warn("Failed to get nodes report ", e);
      }
    }

    // Delete duplicate from all the node reports got from all the available
    // YARN RMs. Nodes can be moved from one subclusters to another. In this
    // operation they result LOST/RUNNING in the previous SubCluster and
    // NEW/RUNNING in the new one.

    return RouterWebServiceUtil.deleteDuplicateNodesInfo(nodes.getNodes());
  }

}