private void onResponses(List<ResponseInfo> responseInfos, Map<Integer, RequestInfo> correlationIdToRequest,
      Map<Integer, RemoteReplicaGroup> correlationIdToReplicaGroup) {
    logger.trace("Thread Name: {} There are {} Responses to handle", threadName, responseInfos.size());
    for (ResponseInfo responseInfo : responseInfos) {
      RequestInfo requestInfo = responseInfo.getRequestInfo();
      if (requestInfo == null) {
        // Only possible when warming up the network client
        DataNodeId dataNodeId = responseInfo.getDataNode();
        responseHandler.onConnectionTimeout(dataNodeId);
      } else {
        int correlationId = requestInfo.getRequest().getCorrelationId();
        // Request comes back, remove this request from the map
        correlationIdToRequest.remove(correlationId);
        RemoteReplicaGroup remoteReplicaGroup = correlationIdToReplicaGroup.remove(correlationId);
        // This correlation id might be removed because the corresponding request timed out. But later we get the response
        // back from the network client.
        if (remoteReplicaGroup != null) {
          RequestOrResponseType type = ((RequestOrResponse) requestInfo.getRequest()).getRequestType();
          logger.debug("Remote node: {} Thread name: {} RemoteReplicaGroup {} Handling response of type {} for {}",
              remoteReplicaGroup.getRemoteDataNode(), threadName, remoteReplicaGroup.getId(), type, correlationId);
          switch (type) {
            case ReplicaMetadataRequest:
              remoteReplicaGroup.handleReplicaMetadataResponse(responseInfo);
              afterHandleReplicaMetadataResponse(remoteReplicaGroup);
              break;
            case GetRequest:
              remoteReplicaGroup.handleGetResponse(responseInfo);
              afterHandleGetResponse(remoteReplicaGroup);
              break;
            default:
              logger.error(
                  "Remote node: {} Thread name: {} RemoteReplicaGroup {} Unexpected type: {} received, discarding",
                  remoteReplicaGroup.getRemoteDataNode(), threadName, remoteReplicaGroup.getId(), type);
          }
        } else {
          logger.trace("Thread Name: {} No RemoteReplicaGroup found for correlation Id {}, it might already time out",
              threadName, correlationId);
        }
      }
    }
    // Release all the ResponseInfo
    responseInfos.forEach(ResponseInfo::release);
  }
Handle the response from the nonblocking network client.
@param responseInfos The list of {@link ResponseInfo}s.
@param correlationIdToRequest The map from correlation id to request.
@param correlationIdToReplicaGroup The map from correlation id to remote replica group.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java