class FederationInterceptorREST extends AbstractRESTRequestInterceptor {

  @Override
  public NodeInfo getNode(String nodeId) {
    Map<SubClusterId, SubClusterInfo> subClustersActive = null;
    try {
      subClustersActive = federationFacade.getSubClusters(true);
    } catch (YarnException e) {
      throw new NotFoundException(e.getMessage());
    }

    if (subClustersActive.isEmpty()) {
      throw new NotFoundException(
          FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE);
    }

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

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

    // Collect all the responses in parallel
    NodeInfo nodeInfo = null;
    for (int i = 0; i < subClustersActive.size(); i++) {
      try {
        Future<NodeInfo> future = compSvc.take();
        NodeInfo nodeResponse = future.get();

        // Check if the node was found in this SubCluster
        if (nodeResponse != null) {
          // Check if the node was already found in a different SubCluster and
          // it has an old health report
          if (nodeInfo == null || nodeInfo.getLastHealthUpdate() <
              nodeResponse.getLastHealthUpdate()) {
            nodeInfo = nodeResponse;
          }
        }
      } catch (Throwable e) {
        LOG.warn("Failed to get node report ", e);
      }
    }
    if (nodeInfo == null) {
      throw new NotFoundException("nodeId, " + nodeId + ", is not found");
    }
    return nodeInfo;
  }

}