private Pair<ServerErrorCode, MessageInfo> getTombStoneFromRemoteReplica(BlobId blobId, String remoteHostName,
      int remoteHostPort) {
    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;
    final String clientId = ON_DEMAND_REPLICATION_CLIENTID_PREFIX + currentNode.getHostname();

    DataInputStream stream = null;
    try {
      // Get the index entries from the remote replica with AdminRequest
      int correlationId = 1;
      AdminRequest adminRequest = new AdminRequest(AdminRequestOrResponseType.BlobIndex, null, correlationId, clientId);
      BlobIndexAdminRequest blobIndexRequest = new BlobIndexAdminRequest(blobId, adminRequest);

      ConnectedChannel connectedChannel =
          connectionPool.checkOutConnection(remoteHostName, remoteDataNode.getPortToConnectTo(),
              connectionPoolCheckoutTimeoutMs);
      ChannelOutput channelOutput = connectedChannel.sendAndReceive(blobIndexRequest);
      stream = channelOutput.getInputStream();
      AdminResponseWithContent adminResponse = AdminResponseWithContent.readFrom(stream);
      if (adminResponse.getError() != ServerErrorCode.NoError) {
        logger.error("ReplicateBlobRequest failed to get tombstone {} from the remote node {} {} {}", blobId,
            remoteHostName, remoteHostPort, adminResponse.getError());
        return new Pair(adminResponse.getError(), null);
      }

      byte[] jsonBytes = adminResponse.getContent();
      Map<String, MessageInfo> messages =
          objectMapper.readValue(jsonBytes, new TypeReference<Map<String, MessageInfo>>() {
          });
      if (messages == null || messages.size() != 1) {
        logger.error("ReplicateBlobRequest adminRequest for {} from the remote node {} {} return {} entries {}", blobId,
            remoteHostName, remoteHostPort, messages == null ? 0 : messages.size(), messages);
        return new Pair<>(ServerErrorCode.UnknownError, null);
      }
      MessageInfo info = messages.values().stream().findFirst().get();
      if (info.isDeleted() != true) {
        logger.error("ReplicateBlobRequest adminRequest {} from {} {} returned unexpected entry {}", blobId,
            remoteHostName, remoteHostPort, info);
        return new Pair<>(ServerErrorCode.UnknownError, null);
      }

      return new Pair(ServerErrorCode.NoError, info);
    } catch (Exception e) { // catch the getBlob exception
      logger.error("ReplicateBlobRequest getTombStoneFromRemoteReplica {} from the remote node {} hit exception ",
          blobId, remoteHostName, e);
      return new Pair(ServerErrorCode.UnknownError, null);
    } finally {
      if (stream != null && stream instanceof NettyByteBufDataInputStream) {
        // if the InputStream is NettyByteBufDataInputStream based, it's time to release its buffer.
        ((NettyByteBufDataInputStream) stream).getBuffer().release();
      }
    }
  }
Get the Tombstone index entry from the remote replica
@param blobId the blob id which we want to get the tombstone record from
@param remoteHostName the host name of the remote replica
@param remoteHostPort the port of the remote replica
@return a pair of the {@link ServerErrorCode} and the tombstone {@link MessageInfo}.
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java