@Override
  public Set<Long> getPartitionsNeedRepair(String sourceHostName, int sourceHostPort, List<Long> partitions)
      throws SQLException {
    long startTime = time.milliseconds();
    String partitionsStr = partitions.stream().map(n -> n.toString()).collect(Collectors.joining(","));
    String query = String.format(GET_PARTITIONS_QUERY_EXCLUDE_SOURCE_REPLICA_TEMPLATE + " (%s) ", partitionsStr);

    try (Connection connection = dataSource.getConnection()) {
      try (PreparedStatement statement = connection.prepareStatement(query)) {
        statement.setShort(1, (short) OperationType.TtlUpdateRequest.ordinal());
        statement.setString(2, sourceHostName);
        statement.setInt(3, sourceHostPort);
        try (ResultSet resultSet = statement.executeQuery()) {
          Set<Long> result = new HashSet<>();
          while (resultSet.next()) {
            Long p = resultSet.getLong(1);
            result.add(p);
          }
          metrics.repairDbGetPartitionTimeInMs.update(time.milliseconds() - startTime);
          return result;
        }
      }
    } catch (SQLException e) {
      metrics.repairDbErrorGetCount.inc();
      logger.error("failed to get records from {} due to {}", dataSource, e.getMessage());
      throw e;
    }
  }
Get the partitions which have TtlUpdate requests to repair.
@param sourceHostName the host name of the source replica
@param sourceHostPort the host port of the source replica
@param partitions the partitions to check
@return the partitions which have TtlUpdate requests to repair.
ambry-mysql/src/main/java/com/github/ambry/repair/MysqlRepairRequestsDb.java