public void handleGetResponse(ResponseInfo responseInfo) {
      if (state != ReplicaGroupReplicationState.GET_REQUEST_SENT) {
        logger.error("Remote node: {} Thread name: {} GetResponse comes back after wrong state {}", remoteDataNode,
            threadName, state);
        return;
      }
      NetworkClientErrorCode networkClientErrorCode = responseInfo.getError();
      logger.trace(
          "Remote node: {} Thread name: {} RemoteReplicaGroup {} GetResponse come back for correlation id {}, NetworkClientError {}",
          remoteDataNode, threadName, id, responseInfo.getRequestInfo().getRequest().getCorrelationId(),
          networkClientErrorCode);
      if (networkClientErrorCode == null) {
        replicationMetrics.updateGetRequestTime(time.milliseconds() - getRequestStartTimeMs, replicatingFromRemoteColo,
            replicatingOverSsl, datacenterName, isNonProgressStandbyReplicaGroup);
        try {
          // Deserialize the request from the given ResponseInfo
          DataInputStream dis = new NettyByteBufDataInputStream(responseInfo.content());
          GetResponse response = GetResponse.readFrom(dis, clusterMap);
          if (response.getError() != ServerErrorCode.NoError) {
            // Getting error for entire response, something unexpected happened.
            replicationMetrics.incrementResponseErrorCount(replicatingFromRemoteColo, datacenterName);
            logger.error("Remote node: {} Thread name: {} RemoteReplicaGroup {} ServerError for GetResponse {}",
                remoteDataNode, threadName, id, response.getError());
            throw new ReplicationException("GetResponse unexpected error " + response.getError(), response.getError());
          }
          ReplicaThread.this.handleGetResponse(response, remoteReplicaInfosToSend,
              exchangeMetadataResponseListToProcess, remoteDataNode, isNonProgressStandbyReplicaGroup);
          logger.trace(
              "Remote node: {} Thread name: {} RemoteReplicaGroup {} Handled GetResponse for correlation id {}",
              remoteDataNode, threadName, id, responseInfo.getRequestInfo().getRequest().getCorrelationId());
          long fixMissingStoreKeysTime = time.milliseconds() - fixMissingStoreKeysStartTimeInMs;
          replicationMetrics.updateFixMissingStoreKeysTime(fixMissingStoreKeysTime, replicatingFromRemoteColo,
              replicatingOverSsl, datacenterName);
        } catch (Exception e) {
          setException(e, "GetResponse unexpected error");
        }
      } else {
        if (networkClientErrorCode != NetworkClientErrorCode.TimeoutError) {
          replicationMetrics.incrementRequestNetworkErrorCount(replicatingFromRemoteColo, datacenterName);
        }
        remoteReplicaInfos.forEach(r -> responseHandler.onEvent(r.getReplicaId(), networkClientErrorCode));
        setException(new IOException("NetworkClientErrorCode: " + networkClientErrorCode), "Failed to send GetRequest");
      }
      state = ReplicaGroupReplicationState.DONE;
    }
Handle the {@link GetResponse} returned in {@link ResponseInfo} by nonblocking network client.
@param responseInfo The {@link ResponseInfo} that carries a {@link GetResponse}.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java