public void handleReplicaMetadataResponse(ResponseInfo responseInfo) {
      if (state != ReplicaGroupReplicationState.REPLICA_METADATA_REQUEST_SENT) {
        logger.error("Remote node: {} Thread name: {} ReplicaMetadataResponse comes back after wrong state {}",
            remoteDataNode, threadName, state);
        return;
      }

      NetworkClientErrorCode networkClientErrorCode = responseInfo.getError();
      logger.trace(
          "Remote node: {} Thread name: {} RemoteReplicaGroup {} ReplicaMetadataResponse come back for correlation id {}, NetworkClientError {}",
          remoteDataNode, threadName, id, responseInfo.getRequestInfo().getRequest().getCorrelationId(),
          networkClientErrorCode);
      if (networkClientErrorCode == null) {
        replicationMetrics.updateMetadataRequestTime(time.milliseconds() - replicaMetadataRequestStartTimeMs,
            replicatingFromRemoteColo, replicatingOverSsl, datacenterName);
        try {
          // Deserialize the request from the given ResponseInfo
          DataInputStream dis = new NettyByteBufDataInputStream(responseInfo.content());
          ReplicaMetadataResponse response = ReplicaMetadataResponse.readFrom(dis, findTokenHelper, clusterMap);
          if (response.getError() != ServerErrorCode.NoError
              || response.getReplicaMetadataResponseInfoList().size() != remoteReplicaInfos.size()) {
            int replicaMetadataResponseInfoListSize = response.getReplicaMetadataResponseInfoList() == null ? 0
                : response.getReplicaMetadataResponseInfoList().size();
            replicationMetrics.incrementResponseErrorCount(replicatingFromRemoteColo, datacenterName);
            exchangeMetadataResponseList = Collections.emptyList();
            // Getting error for entire response, something unexpected happened.
            logger.error(
                "Remote node: {} Thread name: {} RemoteReplicaGroup {} ServerError for ReplicaMetadataResponse {} "
                    + "ReplicaMetadataResponseInfoListSize: {} RemoteReplicaInfoSize: {}", remoteDataNode, threadName,
                id, response.getError(), replicaMetadataResponseInfoListSize, remoteReplicaInfos.size());
            throw new ReplicationException("ReplicaMetadataResponse unexpected error " + response.getError(),
                response.getError());
          }
          exchangeMetadataResponseList =
              ReplicaThread.this.handleReplicaMetadataResponse(response, remoteReplicaInfos, remoteDataNode, storeKeysConversionLog);
          state = ReplicaGroupReplicationState.REPLICA_METADATA_RESPONSE_HANDLED;
          logger.trace(
              "Remote node: {} Thread name: {} RemoteReplicaGroup {} Handled ReplicaMetadataResponse for correlation id {}, set state to {}",
              remoteDataNode, threadName, id, responseInfo.getRequestInfo().getRequest().getCorrelationId(), state);
          long exchangeMetadataTimeInMs = time.milliseconds() - exchangeMetadataStartTimeInMs;
          replicationMetrics.updateExchangeMetadataTime(exchangeMetadataTimeInMs, replicatingFromRemoteColo,
              replicatingOverSsl, datacenterName);
        } catch (Exception e) {
          setException(e, "ReplicaMetadataResponse unexpected error");
        }
      } else {
        // We have a network client error here, mark all the replicas int request with this error.
        if (networkClientErrorCode != NetworkClientErrorCode.TimeoutError) {
          replicationMetrics.incrementRequestNetworkErrorCount(replicatingFromRemoteColo, datacenterName);
        }
        exchangeMetadataResponseList = Collections.emptyList();
        remoteReplicaInfos.forEach(r -> responseHandler.onEvent(r.getReplicaId(), networkClientErrorCode));
        setException(new IOException("NetworkClientErrorCode: " + networkClientErrorCode),
            "Failed to send ReplicaMetadataRequest");
      }
      if (exchangeMetadataListener != null) {
        remoteReplicaInfos.forEach(
            r -> exchangeMetadataListener.onComplete(r.getReplicaId().getPartitionId(), exception));
      }
    }
Handle the {@link ReplicaMetadataResponse} returned in {@link ResponseInfo} by nonblocking network client.
@param responseInfo The {@link ResponseInfo} that carries a {@link ReplicaMetadataResponse}.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java