static void repairRequestTest(MockCluster cluster, SSLConfig clientSSLConfig, boolean testEncryption,
      MockNotificationSystem notificationSystem) throws Exception {
    List<MockDataNodeId> allNodes = cluster.getAllDataNodes();
    MockClusterMap clusterMap = cluster.getClusterMap();
    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS);

    /*
     * sourceDataNode is the source datanode which successfully acks the requests.
     * targetDataNode is the datanode which was temporarily unavailable and then gets back to fix the requests.
     * thirdDataNode is the third data node which was temporarily unavailable for a while.
     */
    DataNodeId sourceDataNode = allNodes.get(0);
    DataNodeId targetDataNode = allNodes.get(1);
    DataNodeId thirdDataNode = allNodes.get(2);
    PartitionId partitionId1 = partitionIds.get(0);
    PartitionId partitionId2 = partitionIds.get(1);
    ConnectedChannel sourceChannel = null;
    ConnectedChannel targetChannel = null;
    ConnectedChannel thirdChannel = null;

    // Open the db connection and use it to generate repair requests.
    MysqlRepairRequestsDb db = createRepairRequestsConnection(sourceDataNode.getDatacenterName(), cluster.time);

    // open connection channel to all data nodes.
    List<ConnectedChannel> channels = new ArrayList<>();
    for (DataNodeId node : allNodes) {
      Port port = new Port(node.getHttp2Port(), PortType.HTTP2);
      ConnectedChannel channel = getBlockingChannelBasedOnPortType(port, "localhost", null, clientSSLConfig);
      channel.connect();
      channels.add(channel);
      if (channel.getRemoteHost().equals(sourceDataNode.getHostname())
          && channel.getRemotePort() == sourceDataNode.getPortToConnectTo().getPort()) {
        sourceChannel = channel;
      } else if (channel.getRemoteHost().equals(targetDataNode.getHostname())
          && channel.getRemotePort() == targetDataNode.getPortToConnectTo().getPort()) {
        targetChannel = channel;
      } else if (channel.getRemoteHost().equals(thirdDataNode.getHostname())
          && channel.getRemotePort() == thirdDataNode.getPortToConnectTo().getPort()) {
        thirdChannel = channel;
      }
    }

    // stop replication on all data nodes
    controlReplicationForPartition(channels, partitionIds, false);
    // stop blobs store on all data nodes except for the sourceChannel and the targetChannel. So we can monitor the ODR on the target node.
    for (ConnectedChannel channel : channels) {
      if (channel == sourceChannel || channel == targetChannel) {
        continue;
      }
      for (PartitionId id : partitionIds) {
        controlBlobStore(channel, id, false);
      }
    }

    BlobIdFactory blobIdFactory = new BlobIdFactory(clusterMap);
    short dataSize = 31870;
    byte[] userMetadata = new byte[1000];
    byte[] data = new byte[dataSize];
    byte[] encryptionKey = new byte[100];
    short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    TestUtils.RANDOM.nextBytes(userMetadata);
    TestUtils.RANDOM.nextBytes(data);
    TestUtils.RANDOM.nextBytes(encryptionKey);

    // property with 7days expiration
    BlobProperties propertiesWithTtl =
        new BlobProperties(dataSize, "serviceid1", "ownerid", "image/png", false, TestUtils.TTL_SECS,
            cluster.time.milliseconds(), accountId, containerId, testEncryption, null, null, null, null);
    long ttlUpdateBlobExpiryTimeMs = getExpiryTimeMs(propertiesWithTtl);
    // property of blob with the infinite TTL
    BlobProperties properties =
        new BlobProperties(dataSize, "serviceid1", accountId, containerId, testEncryption, cluster.time.milliseconds());
    cluster.time.sleep(1000);

    short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();
    BlobId blobId1 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
        properties.getAccountId(), properties.getContainerId(), partitionId1, testEncryption,
        BlobId.BlobDataType.DATACHUNK);
    BlobId blobId2 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
        properties.getAccountId(), properties.getContainerId(), partitionId1, testEncryption,
        BlobId.BlobDataType.DATACHUNK);
    BlobId blobId3 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
        properties.getAccountId(), properties.getContainerId(), partitionId1, testEncryption,
        BlobId.BlobDataType.DATACHUNK);
    BlobId expiredId = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
        properties.getAccountId(), properties.getContainerId(), partitionId2, testEncryption,
        BlobId.BlobDataType.DATACHUNK);
    int numOfRecordsCannotFix = 15; // have set query pagination count to 7.
    int numOfSuccessFix = 27;
    int randomTestNum = 39;

    BlobId blobId;
    long operationTime;
    PutRequest putRequest;

    try {
      /*
       * Stage 1: Test the correctness of the ReplicateBlob. Replicate blobs from sourceDataNode to the targetDataNode
       * On the sourceDataNode and the targetDataNode, the blob is under different states.
       */
      //
      // sourceDataNode has PutBlob and TtlUpdate
      //
      // 1. On the sourceDataNode, the blob is ttlUpdated. On the targetDataNode, it doesn't exist.
      // Add TtlUpdate RepairRequest to the db.
      // ReplicateBlob will replicate both the putBlob and the TtlUpdate
      blobId = blobId1;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, false, ttlUpdateBlobExpiryTimeMs);
      cluster.time.sleep(1000);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, operationTime);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      addRepairRequestForTtlUpdate(db, sourceDataNode, blobId, operationTime, Utils.Infinite_Time);
      // wait for the replication to catch up
      notificationSystem.awaitBlobReplicates(blobId.getID());
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // the TtlUpdate Operation Time will be same as the PutBlob time since we apply TtlUpdate from the GetRequest.
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, propertiesWithTtl.getCreationTimeInMs(),
          Utils.Infinite_Time);
      cluster.time.sleep(1000);

      // 2. On the sourceDataNode, the blob is deleted. On the targetDataNode, it doesn't exist.
      // Add Delete RepairRequest to the db.
      // ReplicateBlob will replicate both the putBlob and the delete
      blobId = blobId2;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, false, ttlUpdateBlobExpiryTimeMs);
      cluster.time.sleep(1000);
      operationTime = cluster.time.milliseconds();
      deleteBlob(sourceChannel, blobId, operationTime, ServerErrorCode.NoError);
      addRepairRequestForDelete(db, sourceDataNode, blobId, operationTime);
      // wait for the replication to catch up
      notificationSystem.awaitBlobReplicates(blobId.getID());
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      checkDeleteRecord(targetChannel, clusterMap, blobId, propertiesWithTtl.getCreationTimeInMs());
      cluster.time.sleep(1000);

      // 3. source doesn't have the PutBlob. Add Delete RepairRequest to the db.
      // ODR will repair the tombstone.
      blobId = blobId3;
      getBlobAndVerify(blobId, sourceChannel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      cluster.time.sleep(1000);
      operationTime = cluster.time.milliseconds();
      addRepairRequestForDelete(db, sourceDataNode, blobId, operationTime);
      // wait for the replication to catch up
      notificationSystem.awaitBlobReplicates(blobId.getID());
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      checkDeleteRecord(targetChannel, clusterMap, blobId, operationTime);
      cluster.time.sleep(1000);

      // 4. On the sourceDataNode, the blob is ttlUpdated. On the targetDataNode, it has the blob.
      // When the Repair starts, the blob is expired.
      // But ODR inherits the original operation time when it was first called. So BlobStore.updateTtl won't reject it.
      blobId = expiredId;
      // propertiesWithExpiredTtl will expire is 2s.
      BlobProperties propertiesWithExpiredTtl =
          new BlobProperties(dataSize, "serviceid1", "ownerid", "image/png", false, TimeUnit.SECONDS.toSeconds(2),
              cluster.time.milliseconds(), accountId, containerId, testEncryption, null, null, null, null);
      // write to both source and target nodes.
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithExpiredTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithExpiredTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      long expiryTimeMs = getExpiryTimeMs(propertiesWithExpiredTtl);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, false, expiryTimeMs);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, false, expiryTimeMs);
      // now after 1s, update Ttl on the source. It is successful.
      cluster.time.sleep(1000);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, operationTime);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // make the blob expired
      cluster.time.sleep(10000);
      // if we update TtlUpdate right now on target with current operation time, it'll fail.
      updateBlobTtl(targetChannel, blobId, cluster.time.milliseconds(), ServerErrorCode.BlobUpdateNotAllowed);
      // But the repairRequest inherit the operation time from the source replica when the request was first sent.
      // so it should be successful.
      addRepairRequestForTtlUpdate(db, sourceDataNode, blobId, operationTime, Utils.Infinite_Time);

      // wait for the replication to catch up
      notificationSystem.awaitBlobReplicates(blobId.getID());
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithExpiredTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, operationTime, Utils.Infinite_Time);
      cluster.time.sleep(1000);

      // Test pagination with records which cannot be fixed as well as records can be fixed.
      // Inject records can not be fixed.
      for (int i = 0; i < numOfRecordsCannotFix; i++) {
        blobId = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
            properties.getAccountId(), properties.getContainerId(), partitionId1, testEncryption,
            BlobId.BlobDataType.DATACHUNK);
        operationTime = cluster.time.milliseconds();
        // will fail to repair. we emit the metrics and repair continues to run for other requests
        addRepairRequestForTtlUpdate(db, sourceDataNode, blobId, operationTime, Utils.Infinite_Time);
        cluster.time.sleep(1);
      }
      // Inject records can be fixed.
      for (int i = 0; i < numOfSuccessFix; i++) {
        blobId = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
            properties.getAccountId(), properties.getContainerId(), partitionId1, testEncryption,
            BlobId.BlobDataType.DATACHUNK);
        operationTime = cluster.time.milliseconds();
        putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
            Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
            testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
        putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
        addRepairRequestForDelete(db, sourceDataNode, blobId, operationTime);
        cluster.time.sleep(1);
      }
      // Wait for good records are fixed.
      DataSource dataSource = db.getDataSource();
      long numberOfRows = 0;
      String rowQuerySql = "SELECT COUNT(*) as total FROM ambry_repair_requests";
      do {
        try (Connection connection = dataSource.getConnection()) {
          try (PreparedStatement statement = connection.prepareStatement(rowQuerySql)) {
            try (ResultSet result = statement.executeQuery()) {
              while (result.next()) {
                numberOfRows = result.getLong("total");
              }
            }
          }
        }
      } while (numberOfRows != numOfRecordsCannotFix);

      /*
       * Stage 2: Start the third data node. Now three data nodes are running.
       * Randomly generate failure requests. Each node will pick up some RepairRequests and fix them.
       */
      for (PartitionId id : partitionIds) {
        controlBlobStore(thirdChannel, id, true);
      }
      List<BlobId> blobIds = new ArrayList<>();
      for (int i = 0; i < randomTestNum; i++) {
        PartitionId partitionId = TestUtils.RANDOM.nextInt(2) == 0 ? partitionId1 : partitionId2;
        blobId = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
            propertiesWithTtl.getAccountId(), propertiesWithTtl.getContainerId(), partitionId, testEncryption,
            BlobId.BlobDataType.SIMPLE);
        putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
            Unpooled.wrappedBuffer(data), propertiesWithTtl.getBlobSize(), BlobType.DataBlob,
            testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
        int successOne = TestUtils.RANDOM.nextInt(3);
        ConnectedChannel successChannel =
            successOne == 0 ? sourceChannel : (successOne == 1 ? targetChannel : thirdChannel);
        DataNodeId successNode = successOne == 0 ? sourceDataNode : (successOne == 1 ? targetDataNode : thirdDataNode);
        putBlob(putRequest, successChannel, ServerErrorCode.NoError);
        cluster.time.sleep(1000);
        operationTime = cluster.time.milliseconds();
        if (TestUtils.RANDOM.nextInt(2) == 0) {
          deleteBlob(successChannel, blobId, operationTime, ServerErrorCode.NoError);
          addRepairRequestForDelete(db, successNode, blobId, operationTime);
        } else {
          updateBlobTtl(successChannel, blobId, operationTime);
          addRepairRequestForTtlUpdate(db, successNode, blobId, operationTime, Utils.Infinite_Time);
        }
        blobIds.add(blobId);
        cluster.time.sleep(1000);
      }
      for (BlobId id : blobIds) {
        notificationSystem.awaitBlobReplicates(id.getID());
      }

      // although we check the notification. DB update is in the RepairRequestsSender not in the handler threads.
      // wait for some time more.
      // We should still have one record which is case 4 error case.
      do {
        try (Connection connection = dataSource.getConnection()) {
          try (PreparedStatement statement = connection.prepareStatement(rowQuerySql)) {
            try (ResultSet result = statement.executeQuery()) {
              while (result.next()) {
                numberOfRows = result.getLong("total");
              }
            }
          }
        }
      } while (numberOfRows != numOfRecordsCannotFix);

      resetRepairRequestsDb(db);

      // Test is done.
      // restart the blobs stores
      for (ConnectedChannel channel : channels) {
        if (channel == sourceChannel || channel == targetChannel || channel == thirdChannel) {
          continue;
        }
        for (PartitionId id : partitionIds) {
          controlBlobStore(channel, id, true);
        }
      }
      // now restart replication on all data nodes
      controlReplicationForPartition(channels, partitionIds, true);
      // close all the connections
      for (ConnectedChannel channel : channels) {
        channel.disconnect();
      }
    } catch (Exception e) {
      e.printStackTrace();
      assertNull(e);
    } finally {
      List<? extends ReplicaId> replicaIds = cluster.getClusterMap()
          .getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS)
          .get(0)
          .getReplicaIds();
      for (ReplicaId replicaId : replicaIds) {
        MockReplicaId mockReplicaId = (MockReplicaId) replicaId;
        ((MockDiskId) mockReplicaId.getDiskId()).setDiskState(HardwareState.AVAILABLE, true);
      }
    }
  }
Test the background RepairRequestsSender and the repair requests handlers.
ambry-test-utils/src/main/java/com/github/ambry/server/ServerTestUtil.java