@Override
  public List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop,
      int pollTimeoutMs) {
    // prepare lists of requests to send and drop
    EnumMap<ReplicaType, List<RequestInfo>> requestsToSendByType = new EnumMap<>(ReplicaType.class);
    EnumMap<ReplicaType, Set<Integer>> requestsToDropByType = new EnumMap<>(ReplicaType.class);
    for (ReplicaType replicaType : childNetworkClients.keySet()) {
      requestsToSendByType.put(replicaType, new ArrayList<>());
      requestsToDropByType.put(replicaType, new HashSet<>());
    }
    for (RequestInfo requestInfo : allRequestsToSend) {
      ReplicaType replicaType = requestInfo.getReplicaId().getReplicaType();
      List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType);
      if (requestsToSend == null) {
        throw new IllegalStateException("No NetworkClient configured for replica type: " + replicaType);
      }
      requestsToSend.add(requestInfo);
      correlationIdToReplicaType.put(requestInfo.getRequest().getCorrelationId(), replicaType);
    }
    for (Integer correlationId : allRequestsToDrop) {
      ReplicaType replicaType = correlationIdToReplicaType.get(correlationId);
      if (replicaType != null) {
        requestsToDropByType.get(replicaType).add(correlationId);
      }
    }

    // send requests using child clients from background threads so that inactive clients do not block the other client
    // from making progress.
    AtomicBoolean wakeupCalled = new AtomicBoolean(false);
    ArrayList<Future<List<ResponseInfo>>> sendAndPollFutures = new ArrayList<>(childNetworkClients.size());
    childNetworkClients.forEach((replicaType, client) -> {
      List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType);
      Set<Integer> requestsToDrop = requestsToDropByType.get(replicaType);
      if (!requestsToSend.isEmpty() || !requestsToDrop.isEmpty()) {
        logger.trace("replicaType={}, requestsToSend={}, requestsToDrop={}", replicaType, requestsToSend,
            requestsToDrop);
      }
      sendAndPollFutures.add(executor.submit(() -> {
        List<ResponseInfo> childClientResponses = client.sendAndPoll(requestsToSend, requestsToDrop, pollTimeoutMs);
        if (wakeupCalled.compareAndSet(false, true)) {
          // the client that gets a response first can wake up the other clients so that they do not waste time waiting
          // for the poll timeout to expire. This helps when one child client is very active and the others have very
          // little activity.
          childNetworkClients.values().stream().filter(c -> c != client).forEach(NetworkClient::wakeup);
        }
        return childClientResponses;
      }));
    });

    // process responses returned by each child client
    List<ResponseInfo> responses = new ArrayList<>();
    for (Future<List<ResponseInfo>> future : sendAndPollFutures) {
      try {
        List<ResponseInfo> responseInfoList = future.get();
        for (ResponseInfo responseInfo : responseInfoList) {
          // clean up correlation ids for completed requests
          if (responseInfo.getRequestInfo() != null) {
            correlationIdToReplicaType.remove(responseInfo.getRequestInfo().getRequest().getCorrelationId());
          }
          responses.add(responseInfo);
        }
      } catch (InterruptedException | ExecutionException e) {
        logger.error("Hit unexpected exception on parallel sendAndPoll.", e);
      }
    }
    return responses;
  }
@param childNetworkClients a map from {@link ReplicaType} to {@link NetworkClient} to use for that type.
ambry-network/src/main/java/com/github/ambry/network/CompositeNetworkClient.java