private Pair<ServerErrorCode, GetResponse> getBlobFromRemoteReplica(ReplicateBlobRequest replicateBlobRequest) {
    BlobId blobId = replicateBlobRequest.getBlobId();
    String remoteHostName = replicateBlobRequest.getSourceHostName();
    int remoteHostPort = replicateBlobRequest.getSourceHostPort();
    DataNodeId remoteDataNode = clusterMap.getDataNodeId(remoteHostName, remoteHostPort);
    if (remoteDataNode == null) {
      logger.error("ReplicateBlobRequest {} couldn't find the remote host {} {} in the clustermap.", blobId,
          remoteHostName, remoteHostPort);
      return new Pair(ServerErrorCode.ReplicaUnavailable, null);
    }

    // ON_DEMAND_REPLICATION_TODO: Add configuration as replicationConfig.replicationConnectionPoolCheckoutTimeoutMs
    final int connectionPoolCheckoutTimeoutMs = 1000;
    List<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
    List<BlobId> blobIds = new ArrayList<>();
    blobIds.add(blobId);
    PartitionRequestInfo partitionInfo = new PartitionRequestInfo(blobId.getPartition(), blobIds);
    partitionRequestInfoList.add(partitionInfo);
    final String clientId = ON_DEMAND_REPLICATION_CLIENTID_PREFIX + currentNode.getHostname();

    try {
      // Get the Blob including expired and deleted.
      GetRequest getRequest = new GetRequest(replicateBlobRequest.getCorrelationId(), clientId, MessageFormatFlags.All,
          partitionRequestInfoList, GetOption.Include_All);

      // get the blob from the remote replica
      ConnectedChannel connectedChannel =
          connectionPool.checkOutConnection(remoteHostName, remoteDataNode.getPortToConnectTo(),
              connectionPoolCheckoutTimeoutMs);
      ChannelOutput channelOutput = connectedChannel.sendAndReceive(getRequest);
      GetResponse getResponse = GetResponse.readFrom(channelOutput.getInputStream(), clusterMap);
      if (getResponse.getError() != ServerErrorCode.NoError) {
        logger.error("ReplicateBlobRequest failed to get blob {} from the remote node {} {} {}", blobId, remoteHostName,
            remoteHostPort, getResponse.getError());
        return new Pair(getResponse.getError(), getResponse);
      }
      if ((getResponse.getPartitionResponseInfoList() == null) || (getResponse.getPartitionResponseInfoList().size()
          == 0)) {
        logger.error("ReplicateBlobRequest {} returned empty list from the remote node {} {} {}", blobId,
            remoteHostName, remoteHostPort, getResponse.getError());
        return new Pair(ServerErrorCode.UnknownError, getResponse);
      }

      // only have one partition. And checked at least it has one entry above.
      PartitionResponseInfo partitionResponseInfo = getResponse.getPartitionResponseInfoList().get(0);
      if (partitionResponseInfo.getErrorCode() != ServerErrorCode.NoError) {
        // the status can be Blob_Deleted or others
        // Since GetOption is Include_All, even it's deleted on the remote replica, we'll still get the PutBlob.
        // One exception is that because of compaction or other reasons, the PutRecord is gone and it returns Blob_Deleted.
        logger.error("ReplicateBlobRequest Blob {} of {} failed with error code {}", blobId, replicateBlobRequest,
            partitionResponseInfo.getErrorCode());
        return new Pair(partitionResponseInfo.getErrorCode(), getResponse);
      }
      List<MessageInfo> messageInfoList = partitionResponseInfo.getMessageInfoList();
      if (messageInfoList == null || messageInfoList.size() != 1) {
        logger.error(
            "ReplicateBlobRequest PartitionResponseInfo response from GetRequest {} {} {} {} {} returned null.",
            partitionResponseInfo, messageInfoList, remoteHostName, remoteHostPort, blobId);
        return new Pair(ServerErrorCode.BlobNotFound, getResponse);
      }

      return new Pair(ServerErrorCode.NoError, getResponse);
    } catch (Exception e) { // catch the getBlob exception
      logger.error("ReplicateBlobRequest getBlob {} from the remote node {} hit exception ", blobId, remoteHostName, e);
      return new Pair(ServerErrorCode.UnknownError, null);
    }
  }
Get the Blob from the remote replica
@param replicateBlobRequest the {@link ReplicateBlobRequest}
@return a pair of the {@link ServerErrorCode} and the {@link GetResponse}.
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java