@Override
  public void handleReplicateBlobRequest(NetworkRequest request) throws IOException, InterruptedException {
    if (connectionPool == null || transformer == null || transformer.get() == null) {
      throw new UnsupportedOperationException("ReplicateBlobRequest is not supported on this node.");
    }

    ReplicateBlobRequest replicateBlobRequest;
    if (request instanceof LocalChannelRequest) {
      // This is a case where handleReplicateBlobRequest is called when frontends are talking to Azure. In this case, this method
      // is called by request handler threads running within the frontend router itself. So, the request can be directly
      // referenced as java objects without any need for deserialization.
      replicateBlobRequest = (ReplicateBlobRequest) ((LocalChannelRequest) request).getRequestInfo().getRequest();
    } else {
      replicateBlobRequest = ReplicateBlobRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap);
    }

    if (replicateBlobRequest.getVersionId() == ReplicateBlobRequest.VERSION_2) {
      handleReplicateBlobRequestV2(request, replicateBlobRequest);
      return;
    }

    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;
    long startTime = 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();
    GetResponse getResponse = null;
    ServerErrorCode errorCode;
    try {
      if (localStoreHasTheKey(replicateBlobRequest)) {
        logger.info("ReplicateBlobRequest replicated Blob {}, local Store has the Key already, do nothing", blobId);
        errorCode = ServerErrorCode.NoError;
      } else {
        // get the Blob from the remote replica.
        Pair<ServerErrorCode, GetResponse> getResult = getBlobFromRemoteReplica(replicateBlobRequest);
        errorCode = getResult.getFirst();
        getResponse = getResult.getSecond();

        if (errorCode == ServerErrorCode.NoError) {
          // getBlobFromRemoteReplicate has checked partitionResponseInfoList's size is 1 and it has one MessageInfo.
          PartitionResponseInfo partitionResponseInfo = getResponse.getPartitionResponseInfoList().get(0);
          List<MessageInfo> messageInfoList = partitionResponseInfo.getMessageInfoList();
          MessageInfo orgMsgInfo = messageInfoList.get(0);

          // Transfer the input stream with transformer
          Message output = MessageSievingInputStream.transferInputStream(Collections.singletonList(transformer.get()),
              getResponse.getInputStream(), orgMsgInfo);
          if (output == null) {
            logger.error("ReplicateBlobRequest transferInputStream {} returned null, {} {} {}", orgMsgInfo,
                remoteHostName, remoteHostPort, blobId);
            errorCode = ServerErrorCode.UnknownError;
          } else {
            // write the message to the local store
            MessageFormatWriteSet writeset =
                new MessageFormatWriteSet(output.getStream(), Collections.singletonList(output.getMessageInfo()),
                    false);
            Store store = storeManager.getStore(blobId.getPartition());
            store.put(writeset);

            // also applyTtlUpdate and applyDelete if needed.
            // the operation time of TtlUpdate and Delete will be same as the source replica's PutBlob creation time.
            // But it's ok. Replication thread should have the same behavior.
            // We don't care the operation time of TtlUpdate and Delete that much.
            if (orgMsgInfo.isTtlUpdated()) {
              applyTtlUpdate(orgMsgInfo, replicateBlobRequest);
            }
            if (orgMsgInfo.isDeleted()) {
              applyDelete(orgMsgInfo, replicateBlobRequest);
            }
            logger.info("ReplicateBlobRequest replicated Blob {} from remote host {} {}", blobId, remoteHostName,
                remoteHostPort);
            errorCode = ServerErrorCode.NoError;
          } // if (output == null)
        } else if (errorCode == ServerErrorCode.BlobDeleted) {
          if (serverConfig.serverReplicateTombstoneEnabled) {
            // If GetBlob with GetOption.Include_All returns Blob_Deleted.
            // it means the remote peer probably only have a delete tombstone for this blob.
            // get the tombstone index entry from the remote replica and force write the delete record.
            metrics.replicateDeleteRecordRate.mark();
            Pair<ServerErrorCode, MessageInfo> indexEntryResult =
                getTombStoneFromRemoteReplica(blobId, remoteHostName, remoteHostPort);
            errorCode = indexEntryResult.getFirst();
            MessageInfo indexInfo = indexEntryResult.getSecond();

            if (errorCode == ServerErrorCode.NoError) {
              Store store = storeManager.getStore(blobId.getPartition());
              // if forceDelete fail due to local store has the key, it throws Already_Exist exception
              store.forceDelete(Collections.singletonList(indexInfo));
              logger.info("ReplicateBlobRequest replicated tombstone {} from remote host {} {} : {}", blobId,
                  remoteHostName, remoteHostPort, indexInfo);
            }
          } // if (serverConfig.serverReplicateTombstoneEnabled)
        } // if (errorCode == XXX)
      } // if (remoteDataNode.equals(currentNode))
    } catch (StoreException e) { // catch the store write exception
      if (e.getErrorCode() == StoreErrorCodes.AlreadyExist) {
        logger.info("ReplicateBlobRequest Blob {} already exists for {}", blobId, replicateBlobRequest);
        errorCode = ServerErrorCode.NoError;
      } else {
        logger.error("ReplicateBlobRequest unknown exception to replicate {} of {}", blobId, replicateBlobRequest, e);
        errorCode = ErrorMapping.getStoreErrorMapping(e.getErrorCode());
      }
    } catch (Exception e) {
      // localStoreHasTheKey calls getConvertedStoreKeys which may throw Exception
      logger.error("ReplicateBlobRequest unknown exception to replicate {} of {}", blobId, replicateBlobRequest, e);
      errorCode = ServerErrorCode.UnknownError;
    } finally {
      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();
      }
    }

    ReplicateBlobResponse response =
        new ReplicateBlobResponse(replicateBlobRequest.getCorrelationId(), replicateBlobRequest.getClientId(),
            errorCode);
    long processingTime = SystemTime.getInstance().milliseconds() - startTime;
    totalTimeSpent += processingTime;
    publicAccessLogger.info("{} {} processingTime {}", replicateBlobRequest, response, processingTime);
    // Update request metrics.
    RequestMetricsUpdater metricsUpdater = new RequestMetricsUpdater(requestQueueTime, processingTime, 0, 0, false);
    replicateBlobRequest.accept(metricsUpdater);
    requestResponseChannel.sendResponse(response, request,
        new ServerNetworkResponseMetrics(metrics.replicateBlobResponseQueueTimeInMs, metrics.replicateBlobSendTimeInMs,
            metrics.replicateBlobTotalTimeInMs, null, null, totalTimeSpent));
  }
If the replicateBlob is replicating from this node or the local store has the key already, return true.
@param replicateBlobRequest the {@link ReplicateBlobRequest}
@return true if replicates from this node or the local store has the key
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java