private <T extends Message> void groupCall(CompletableFuture<T> future, Set<ServerName> servers,
    List<ClientMetaService.Interface> stubs, int startIndexInclusive, Callable<T> callable,
    Predicate<T> isValidResp, String debug, ConcurrentLinkedQueue<Throwable> errors) {
    int endIndexExclusive = Math.min(startIndexInclusive + hedgedReadFanOut, stubs.size());
    AtomicInteger remaining = new AtomicInteger(endIndexExclusive - startIndexInclusive);
    for (int i = startIndexInclusive; i < endIndexExclusive; i++) {
      addListener(call(stubs.get(i), callable), (r, e) -> {
        // a simple check to skip all the later operations earlier
        if (future.isDone()) {
          return;
        }
        if (e == null && !isValidResp.test(r)) {
          e = badResponse(debug);
        }
        if (e != null) {
          // make sure when remaining reaches 0 we have all exceptions in the errors queue
          errors.add(e);
          if (remaining.decrementAndGet() == 0) {
            if (endIndexExclusive == stubs.size()) {
              // we are done, complete the future with exception
              RetriesExhaustedException ex =
                new RetriesExhaustedException("masters", stubs.size(), new ArrayList<>(errors));
              future.completeExceptionally(new MasterRegistryFetchException(servers, ex));
            } else {
              groupCall(future, servers, stubs, endIndexExclusive, callable, isValidResp, debug,
                errors);
            }
          }
        } else {
          // do not need to decrement the counter any more as we have already finished the future.
          future.complete(r);
        }
      });
    }
  }
send requests concurrently to hedgedReadsFanout end points. If any of the request is succeeded,
we will complete the future and quit. If all the requests in one round are failed, we will
start another round to send requests concurrently tohedgedReadsFanout end points. If all end
points have been tried and all of them are failed, we will fail the future.
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractRpcBasedConnectionRegistry.java