@Override
  public void handleReplicaMetadataRequest(NetworkRequest request) throws IOException, InterruptedException {
    if (replicationEngine == null) {
      throw new UnsupportedOperationException("Replication not supported on this node.");
    }
    ReplicaMetadataRequest replicaMetadataRequest =
        ReplicaMetadataRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap, findTokenHelper);
    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;

    List<ReplicaMetadataRequestInfo> replicaMetadataRequestInfoList =
        replicaMetadataRequest.getReplicaMetadataRequestInfoList();
    int partitionCnt = replicaMetadataRequestInfoList.size();
    long startTimeInMs = SystemTime.getInstance().milliseconds();
    ReplicaMetadataResponse response = null;
    try {
      List<ReplicaMetadataResponseInfo> replicaMetadataResponseList = new ArrayList<>(partitionCnt);
      for (ReplicaMetadataRequestInfo replicaMetadataRequestInfo : replicaMetadataRequestInfoList) {
        long partitionStartTimeInMs = SystemTime.getInstance().milliseconds();
        PartitionId partitionId = replicaMetadataRequestInfo.getPartitionId();
        ReplicaType replicaType = replicaMetadataRequestInfo.getReplicaType();
        ServerErrorCode error = validateRequest(partitionId, RequestOrResponseType.ReplicaMetadataRequest, false);
        logger.trace("{} Time used to validate metadata request: {}", partitionId,
            (SystemTime.getInstance().milliseconds() - partitionStartTimeInMs));

        if (error != ServerErrorCode.NoError) {
          logger.error("Validating replica metadata request failed with error {} for partition {}", error, partitionId);
          ReplicaMetadataResponseInfo replicaMetadataResponseInfo =
              new ReplicaMetadataResponseInfo(partitionId, replicaType, error,
                  ReplicaMetadataResponse.getCompatibleResponseVersion(replicaMetadataRequest.getVersionId()));
          replicaMetadataResponseList.add(replicaMetadataResponseInfo);
        } else {
          try {
            FindToken findToken = replicaMetadataRequestInfo.getToken();
            String hostName = replicaMetadataRequestInfo.getHostName();
            String replicaPath = replicaMetadataRequestInfo.getReplicaPath();
            Store store = storeManager.getStore(partitionId);

            partitionStartTimeInMs = SystemTime.getInstance().milliseconds();
            FindInfo findInfo =
                store.findEntriesSince(findToken, replicaMetadataRequest.getMaxTotalSizeOfEntriesInBytes(), hostName,
                    replicaPath);
            logger.trace("{} Time used to find entry since: {}", partitionId,
                (SystemTime.getInstance().milliseconds() - partitionStartTimeInMs));

            // Compute CRC for backup recovery and verification
            if (replicaMetadataRequestInfo.getReplicaPath().startsWith(BackupCheckerThread.DR_Verifier_Keyword)) {
              List<MessageInfo> newMessageInfos = new ArrayList<>();
              // for-each blob
              findInfo.getMessageEntries().stream().forEach(minfo -> {
                MessageInfo newMsgInfo = minfo;
                if (!(minfo.isDeleted() || minfo.isExpired())) {
                  try {
                    Long crc = store.getBlobContentCRC(minfo);
                    newMsgInfo = new MessageInfo(minfo, crc);
                  } catch (Throwable e) {
                    logger.error("Failed to get CRC for blob {} due to {}", minfo.getStoreKey().getID(), e);
                  }
                }
                newMessageInfos.add(newMsgInfo);
              });
              // Create new list of responses
              findInfo = new FindInfo(newMessageInfos, findInfo.getFindToken());
            }


            partitionStartTimeInMs = SystemTime.getInstance().milliseconds();
            long totalBytesRead = findInfo.getFindToken().getBytesRead();
            replicationEngine.updateTotalBytesReadByRemoteReplica(partitionId, hostName, replicaPath, totalBytesRead);
            logger.trace("{} Time used to update total bytes read: {}", partitionId,
                (SystemTime.getInstance().milliseconds() - partitionStartTimeInMs));

            partitionStartTimeInMs = SystemTime.getInstance().milliseconds();
            logger.trace("{} Time used to get remote replica lag in bytes: {}", partitionId,
                (SystemTime.getInstance().milliseconds() - partitionStartTimeInMs));

            ReplicaMetadataResponseInfo replicaMetadataResponseInfo =
                new ReplicaMetadataResponseInfo(partitionId, replicaType, findInfo.getFindToken(),
                    findInfo.getMessageEntries(), getRemoteReplicaLag(store, totalBytesRead),
                    ReplicaMetadataResponse.getCompatibleResponseVersion(replicaMetadataRequest.getVersionId()));
            if (replicaMetadataResponseInfo.getTotalSizeOfAllMessages()
                > 5 * replicaMetadataRequest.getMaxTotalSizeOfEntriesInBytes()) {
              logger.debug("{} generated a metadata response {} where the cumulative size of messages is {}",
                  replicaMetadataRequest, replicaMetadataResponseInfo,
                  replicaMetadataResponseInfo.getTotalSizeOfAllMessages());
              metrics.replicationResponseMessageSizeTooHigh.inc();
            }
            replicaMetadataResponseList.add(replicaMetadataResponseInfo);
            metrics.replicaMetadataTotalSizeOfMessages.update(replicaMetadataResponseInfo.getTotalSizeOfAllMessages());
          } catch (StoreException e) {
            logger.error("Store exception on a replica metadata request with error code {} for partition {}",
                e.getErrorCode(), partitionId, e);
            if (e.getErrorCode() == StoreErrorCodes.IOError) {
              metrics.storeIOError.inc();
            } else {
              metrics.unExpectedStoreFindEntriesError.inc();
            }
            ReplicaMetadataResponseInfo replicaMetadataResponseInfo =
                new ReplicaMetadataResponseInfo(partitionId, replicaType,
                    ErrorMapping.getStoreErrorMapping(e.getErrorCode()),
                    ReplicaMetadataResponse.getCompatibleResponseVersion(replicaMetadataRequest.getVersionId()));
            replicaMetadataResponseList.add(replicaMetadataResponseInfo);
          }
        }
      }
      response =
          new ReplicaMetadataResponse(replicaMetadataRequest.getCorrelationId(), replicaMetadataRequest.getClientId(),
              ServerErrorCode.NoError, replicaMetadataResponseList,
              ReplicaMetadataResponse.getCompatibleResponseVersion(replicaMetadataRequest.getVersionId()));
    } catch (Exception e) {
      logger.error("Unknown exception for request {}", replicaMetadataRequest, e);
      response =
          new ReplicaMetadataResponse(replicaMetadataRequest.getCorrelationId(), replicaMetadataRequest.getClientId(),
              ServerErrorCode.UnknownError,
              ReplicaMetadataResponse.getCompatibleResponseVersion(replicaMetadataRequest.getVersionId()));
    } finally {
      long processingTime = SystemTime.getInstance().milliseconds() - startTimeInMs;
      totalTimeSpent += processingTime;
      publicAccessLogger.info("{} {} processingTime {}", replicaMetadataRequest, response, processingTime);
      logger.trace("{} {} processingTime {}", replicaMetadataRequest, response, processingTime);
      long responseSizeInBytes = response != null ? response.sizeInBytes() : 0L;
      // Update request metrics.
      RequestMetricsUpdater metricsUpdater =
          new RequestMetricsUpdater(requestQueueTime, processingTime, 0, responseSizeInBytes, false);
      replicaMetadataRequest.accept(metricsUpdater);
    }

    requestResponseChannel.sendResponse(response, request,
        new ServerNetworkResponseMetrics(metrics.replicaMetadataResponseQueueTimeInMs,
            metrics.replicaMetadataSendTimeInMs, metrics.replicaMetadataTotalTimeInMs, null, null, totalTimeSpent));
  }
Handler for all metrics to be updated for batch delete operation.
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java