private Pair<List<RequestInfo>, Long> getAmbryRequest(long partitionId, long pageToken) throws Exception {
    List<RequestInfo> requestInfos = new ArrayList<>();
    // get the repair requests for this partition and this node is not the source replica.
    Pair<List<RepairRequestRecord>, Long> records =
        db.getRepairRequestsExcludingHost(partitionId, nodeId.getHostname(), nodeId.getPort(), pageToken);
    Long nextPageToken = records.getSecond();
    for (RepairRequestRecord record : records.getFirst()) {
      BlobId blobId;
      try {
        blobId = new BlobId(record.getBlobId(), clusterMap);
      } catch (IOException e) {
        // new BlobId may generate IOException
        metrics.repairSenderErrorGenerateReqCount.inc();
        logger.error("RepairRequests Sender: Failed to generate Blob ID {}", record, e);
        continue;
      }

      ReplicaId replicaId = partition2Replicas.get(partitionId);
      if (replicaId == null) {
        metrics.repairSenderErrorGenerateReqCount.inc();
        logger.error("RepairRequests Sender: Replica is null {} {} {}", partitionId, record, replicaId);
        continue;
      }

      RequestOrResponseType type;
      if (record.getOperationType() == RepairRequestRecord.OperationType.DeleteRequest) {
        type = RequestOrResponseType.DeleteRequest;
      } else if (record.getOperationType() == RepairRequestRecord.OperationType.TtlUpdateRequest) {
        type = RequestOrResponseType.TtlUpdateRequest;
      } else {
        metrics.repairSenderErrorGenerateReqCount.inc();
        logger.error("RepairRequests Sender: Un-supported repair operation type " + record);
        continue;
      }

      if (record.getSourceHostName().equals(nodeId.getHostname()) && record.getSourceHostPort() == nodeId.getPort()) {
        // shouldn't happen
        String errorMsg =
            "RepairRequests Sender: Shouldn't get the repair requests of which this node is the source replica, "
                + record;
        logger.error(errorMsg);
        metrics.repairSenderErrorGenerateReqCount.inc();
        throw new Exception(errorMsg);
      }

      int correlationId = correlationIdGenerator.incrementAndGet();
      ReplicateBlobRequest request =
          new ReplicateBlobRequest(correlationId, SERVICE_ID, blobId, record.getSourceHostName(),
              record.getSourceHostPort(), type, record.getOperationTimeMs(), record.getLifeVersion(),
              record.getExpirationTimeMs());
      RequestInfo requestInfo =
          new RequestInfo(nodeId.getHostname(), nodeId.getPortToConnectTo(), request, replicaId, null);
      requestInfos.add(requestInfo);
    }

    return new Pair(requestInfos, nextPageToken);
  }
Get the {@link RepairRequestRecord} from the database and generate {@link ReplicateBlobRequest}
@param partitionId the partition id
@param pageToken token to get the page
@return a pair of list of {@link RequestInfo} and token of the next page
ambry-server/src/main/java/com/github/ambry/server/RepairRequestsSender.java