public void handleReplicateBlobRequestV2(NetworkRequest request, ReplicateBlobRequest replicateBlobRequest)
      throws IOException, InterruptedException {
    long startProcessTime = SystemTime.getInstance().milliseconds();
    // Get the parameters from the replicateBlobRequest: the blobId, the remoteHostName and the remoteHostPort.
    BlobId blobId = replicateBlobRequest.getBlobId();
    String remoteHostName = replicateBlobRequest.getSourceHostName();
    int remoteHostPort = replicateBlobRequest.getSourceHostPort();

    // check the request parameters.
    // operationType should be either TtlUpdateRequest or DeleteRequest to fix TtlUpdate record or delete record.
    // For lifeVersion, suppose we should always provide MessageInfo.LIFE_VERSION_FROM_FRONTEND.
    if ((replicateBlobRequest.getOperationType() != RequestOrResponseType.TtlUpdateRequest
        && replicateBlobRequest.getOperationType() != RequestOrResponseType.DeleteRequest) || (
        replicateBlobRequest.getLifeVersion() != MessageInfo.LIFE_VERSION_FROM_FRONTEND)) {
      logger.error("ReplicateBlobRequest invalid request {}", replicateBlobRequest);
      completeReplicateRequest(request, replicateBlobRequest, ServerErrorCode.BadRequest, startProcessTime);
      return;
    }

    // this replica itself is the source replica. Don't need do anything. Return success status.
    DataNodeId remoteDataNode = clusterMap.getDataNodeId(remoteHostName, remoteHostPort);
    if (remoteDataNode != null && remoteDataNode.equals(currentNode)) {
      logger.info("ReplicateBlobRequest this is the source replica, return immediately. {}", replicateBlobRequest);
      completeReplicateRequest(request, replicateBlobRequest, ServerErrorCode.NoError, startProcessTime);
      return;
    }

    PartitionId partitionId = blobId.getPartition();
    ServerErrorCode errorCode = validateRequest(partitionId, RequestOrResponseType.ReplicateBlobRequest, false);
    if (errorCode != ServerErrorCode.NoError) {
      logger.error("Validating ReplicateBlobRequest request failed with error {} for request {}", errorCode,
          replicateBlobRequest);
      completeReplicateRequest(request, replicateBlobRequest, errorCode, startProcessTime);
      return;
    }

    // check if local store has the key already
    MessageInfo localMessageInfo = null;
    try {
      StoreKey convertedKey = getConvertedStoreKeys(Collections.singletonList(blobId)).get(0);
      if (!(convertedKey).equals(blobId)) {
        logger.info("ReplicateBlobRequest replicated Blob {} convert to local key {}", blobId, (BlobId) convertedKey);
        blobId = (BlobId) convertedKey;
      }
      Store store = storeManager.getStore(blobId.getPartition());
      localMessageInfo = store.findKey(convertedKey);
      logger.trace("ReplicateBlobRequest replicated Blob {}, local Store has the Key", blobId);
    } catch (StoreException e) {
      // store.findKey throws e.getErrorCode() == StoreErrorCodes.ID_Not_Found if it doesn't exist.
      if (e.getErrorCode() == StoreErrorCodes.IDNotFound) {
        errorCode = ServerErrorCode.NoError;
      } else {
        logger.error("ReplicateBlobRequest store.findKey throw exception {} on Blob {},", e.getErrorCode(), blobId);
        errorCode = ErrorMapping.getStoreErrorMapping(e.getErrorCode());
      }
    } catch (Exception e) {
      // getConvertedStoreKeys which may throw Exception
      logger.error("ReplicateBlobRequest unknown exception to replicate {} of {}", blobId, replicateBlobRequest, e);
      errorCode = ServerErrorCode.UnknownError;
    }
    if (errorCode != ServerErrorCode.NoError) {
      completeReplicateRequest(request, replicateBlobRequest, errorCode, startProcessTime);
      return;
    }

    if (replicateBlobRequest.getOperationType() == RequestOrResponseType.TtlUpdateRequest) {
      // Repair the PutBlob if local store doesn't have the blob when repair type is RequestOrResponseType.TtlUpdateRequest
      if (localMessageInfo == null) {
        Pair<ServerErrorCode, GetResponse> remoteGetResult = getBlobFromRemoteReplica(replicateBlobRequest);
        errorCode = remoteGetResult.getFirst();
        GetResponse getResponse = remoteGetResult.getSecond();
        if (errorCode == ServerErrorCode.NoError) {
          errorCode = repairPutBlob(replicateBlobRequest, blobId, getResponse);
        }
        if (getResponse != null && getResponse.getInputStream() instanceof NettyByteBufDataInputStream) {
          // if the InputStream is NettyByteBufDataInputStream based, it's time to release its buffer.
          ((NettyByteBufDataInputStream) (getResponse.getInputStream())).getBuffer().release();
        }
      }
      // now repair the TtlUpdate
      if (errorCode == ServerErrorCode.NoError) {
        errorCode = repairTtlUpdate(replicateBlobRequest, blobId);
      }
    } else {
      // replicateBlobRequest.getOperationType() == RequestOrResponseType.DeleteRequest
      if (localMessageInfo != null) {
        // local has the blob, we repair the delete record with information provided by the replicateBlobRequest
        errorCode = repairDeleteRecordToLocalBlob(replicateBlobRequest, blobId);
      } else {
        // local doesn't have the blob,
        // 1. try to replicate the blob from the source replica and repair the delete record.
        // 2. if source doesn't have the PutBlob, write the tombstone to the local store.
        Pair<ServerErrorCode, GetResponse> remoteGetResult = getBlobFromRemoteReplica(replicateBlobRequest);
        errorCode = remoteGetResult.getFirst();
        GetResponse getResponse = remoteGetResult.getSecond();
        if (errorCode == ServerErrorCode.NoError) {
          errorCode = repairPutBlob(replicateBlobRequest, blobId, getResponse);
          if (errorCode == ServerErrorCode.NoError) {
            errorCode = repairDeleteRecordToLocalBlob(replicateBlobRequest, blobId);
          }
        } else if (errorCode == ServerErrorCode.BlobDeleted || errorCode == ServerErrorCode.BlobNotFound) {
          errorCode = repairTombStoneToLocalStore(replicateBlobRequest, blobId);
        }
        if (getResponse != null && getResponse.getInputStream() instanceof NettyByteBufDataInputStream) {
          // if the InputStream is NettyByteBufDataInputStream based, it's time to release its buffer.
          ((NettyByteBufDataInputStream) (getResponse.getInputStream())).getBuffer().release();
        }
      }
    }
    completeReplicateRequest(request, replicateBlobRequest, errorCode, startProcessTime);
  }
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