public void run() {
    while (!shutdown) {
      try {
        boolean hasMore = true;
        boolean hasError = false;
        // if some partitions have more requests, continue to handle them.
        while (hasMore && !hasError && !shutdown) {
          // Based on the RepairRequest status, control the compaction on each partition.
          partitionCompactionControl();

          /*
           * handle the RepairRequests
           */
          hasMore = false;
          hasError = false;
          // loop all the partitions. For each partition, we handle maxResults number of requests.
          for (long partitionId : partitionIds) {
            if (shutdown) {
              break;
            }
            // get the repair requests for this partition
            long oldToken = partition2DbPageToken.get(partitionId);
            Pair<List<RequestInfo>, Long> dbReqs = getAmbryRequest(partitionId, oldToken);
            List<RequestInfo> requestInfos = dbReqs.getFirst();
            long newToken = dbReqs.getSecond();
            partition2DbPageToken.put(partitionId, newToken);

            if (requestInfos.isEmpty()) {
              continue;
            }
            // this partition may have more requests.
            if (requestInfos.size() >= maxResults) {
              hasMore = true;
            }
            logger.info("RepairRequests Sender host {} partition {}, old token {}, new token {} request count {}",
                nodeId, partitionId, oldToken, newToken, requestInfos.size());

            Set<Integer> requestsToDrop = new HashSet<>();
            // send the repair requests to the handler and wait for the responses.
            for (RequestInfo reqInfo : requestInfos) {
              long startTime = System.currentTimeMillis();
              List<ResponseInfo> responses =
                  client.sendAndPoll(Collections.singletonList(reqInfo), requestsToDrop, POLL_TIMEOUT_MS);
              metrics.repairSenderHandleTimeInMs.update(System.currentTimeMillis() - startTime);
              if (responses == null || responses.size() == 0) {
                metrics.repairSenderErrorHandleCount.inc();
                logger.error("RepairRequests Sender: timeout waiting for repairs {} {}", nodeId, reqInfo.getRequest());
                hasError = true;
                break;
              }
              ResponseInfo resInfo = responses.get(0);
              ReplicateBlobResponse res = (ReplicateBlobResponse) resInfo.getResponse();
              ReplicateBlobRequest req = (ReplicateBlobRequest) resInfo.getRequestInfo().getRequest();
              if (res.getError() == ServerErrorCode.NoError) {
                RepairRequestRecord.OperationType type;
                if (req.getOperationType() == RequestOrResponseType.TtlUpdateRequest) {
                  type = RepairRequestRecord.OperationType.TtlUpdateRequest;
                } else {
                  type = RepairRequestRecord.OperationType.DeleteRequest;
                }
                db.removeRepairRequests(req.getBlobId().toString(), type);
                metrics.repairSenderSuccessHandleCount.inc();
                logger.info("RepairRequests Sender: Repaired {}", req);
              } else if ((req.getOperationType() == RequestOrResponseType.TtlUpdateRequest) && (
                  res.getError() == ServerErrorCode.BlobNotFound || res.getError() == ServerErrorCode.BlobExpired)) {
                // Ideally we may need create alert if we cannot repair the requests in timely matter.
                metrics.repairSenderTtlUpdateTooLateCount.inc();
                logger.error("RepairRequests Sender: it's too late to updateTTL {} response {}", req, res);
                hasError = true;
              } else {
                // Ideally we may need create alert if we cannot repair the requests in timely matter.
                metrics.repairSenderErrorHandleCount.inc();
                logger.error("RepairRequests Sender: failed to repair {} response {}", req, res);
                hasError = true;
              }
            }
          } // loop all the partitions
        }  // loop until all the partitions are clean

        // sleep for some time
        if (!shutdown) {
          Thread.sleep(SLEEP_TIME_MS);
        }
      } catch (Throwable e) {
        metrics.repairSenderErrorHandleCount.inc();
        logger.error("RepairRequests Sender: Exception when handling request", e);
      }
    }

    logger.info("RepairRequestsSender thread is closed.");
  }
RepairRequestsSender constructor
@param requestChannel the channel to send requests and receive responses
@param factory the {@link LocalNetworkClientFactory} used to generate the {@link LocalNetworkClient}
@param clusterMap the cluster map
@param nodeId this {@link DataNodeId}
@param repairRequestsDb the database to get the RepairRequestRecord
@param clusterParticipant {@link ClusterParticipant}
@param metricsRegistry the {@link MetricRegistry} to record metrics
ambry-server/src/main/java/com/github/ambry/server/RepairRequestsSender.java