Fix comments

Author: Володин Кирилл Сергеевич

kafka/include/userver/kafka/consumer_scope.hpp

@@ -11,6 +11,7 @@ namespace kafka {
 namespace impl {
 
 class Consumer;
+struct OffsetRange;
 
 }  // namespace impl
 
@@ -118,15 +119,14 @@ class ConsumerScope final {
     /// @warning This is a blocking call
     /// @param topic The name of the Kafka topic.
     /// @param partition The partition number of the Kafka topic.
-    /// @return A pair of integers representing the minimum and maximum offsets for the given topic and partition.
-    ///         The first value is the minimum offset, and the second is the maximum offset.
-    std::pair<std::int64_t, std::int64_t> GetMinMaxOffset(const std::string& topic, std::int32_t partition);
+    /// @return A struct with minimum and maximum offsets for the given topic and partition.
+    OffsetRange GetOffsetRange(const std::string& topic, std::int32_t partition) const;
 
     /// @brief Retrieves the partition IDs for the specified Kafka topic.
     /// @warning This is a blocking call
     /// @param topic The name of the Kafka topic.
     /// @return A vector of partition IDs for the given topic.
-    std::vector<std::int32_t> GetPartitionsId(const std::string& topic);
+    std::vector<std::uint32_t> GetPartitionIds(const std::string& topic) const;
 
 private:
     friend class impl::Consumer;

kafka/include/userver/kafka/impl/consumer.hpp

@@ -18,6 +18,7 @@ class ConsumerImpl;
 
 struct ConsumerConfiguration;
 struct Secret;
+struct OffsetRange;
 
 /// @brief Parameters Consumer uses in runtime.
 /// The struct is used only for documentation purposes, Consumer can be
@@ -96,6 +97,14 @@ class Consumer final {
     /// understanding
     void AsyncCommit();
 
+    /// @brief Retrieves the low and high offsets for the specified kafka topic and partition.
+    /// @see ConsumerScope::GetOffsetRange for better commitment process
+    OffsetRange GetOffsetRange(const std::string& topic, std::int32_t partition) const;
+
+    /// @brief Retrieves the partition IDs for the specified kafka topic.
+    /// @see ConsumerScope::GetPartitionIds for better commitment process
+    std::vector<std::uint32_t> GetPartitionIds(const std::string& topic) const;
+
     /// @brief Adds consumer name to current span.
     void ExtendCurrentSpan() const;
 

kafka/include/userver/kafka/impl/offset.hpp

@@ -0,0 +1,14 @@
+#pragma once
+
+USERVER_NAMESPACE_BEGIN
+
+namespace kafka::impl {
+
+struct OffsetRange final {
+    std::int64_t high{};
+    std::int64_t low{};
+};
+
+}  // namespace kafka::impl
+
+USERVER_NAMESPACE_END

kafka/src/kafka/consumer_scope.cpp

@@ -16,12 +16,12 @@ void ConsumerScope::Stop() noexcept { consumer_.Stop(); }
 
 void ConsumerScope::AsyncCommit() { consumer_.AsyncCommit(); }
 
-std::pair<std::int64_t, std::int64_t> ConsumerScope::GetMinMaxOffset(const std::string& topic, std::int32_t partition) {
-    return consumer_.GetMinMaxOffset(topic, partition);
+OffsetRange ConsumerScope::GetOffsetRange(const std::string& topic, std::int32_t partition) const {
+    return consumer_.GetOffsetRange(topic, partition);
 }
 
-std::vector<std::int32_t> ConsumerScope::GetPartitionsId(const std::string& topic){
-    return consumer_.GetPartitionsId(topic);
+std::vector<std::uint32_t> ConsumerScope::GetPartitionIds(const std::string& topic) const {
+    return consumer_.GetPartitionIds(topic);
 }
 
 }  // namespace kafka

kafka/src/kafka/impl/consumer.cpp

@@ -7,6 +7,7 @@
 #include <userver/engine/sleep.hpp>
 #include <userver/formats/json/value_builder.hpp>
 #include <userver/kafka/impl/configuration.hpp>
+#include <userver/kafka/impl/offset.hpp>
 #include <userver/kafka/impl/stats.hpp>
 #include <userver/testsuite/testpoint.hpp>
 #include <userver/tracing/span.hpp>
@@ -166,6 +167,14 @@ void Consumer::AsyncCommit() {
     }).Get();
 }
 
+OffsetRange Consumer::GetOffsetRange(const std::string& topic, std::int32_t partition) const {
+    consumer_->GetOffsetRange(topic, partition);
+}
+
+std::vector<std::uint32_t> Consumer::GetPartitionIds(const std::string& topic) const {
+    consumer_->GetPartitionIds(topic);
+}
+
 void Consumer::Stop() noexcept {
     if (processing_.exchange(false) && poll_task_.IsValid()) {
         UINVARIANT(consumer_, "Stopping already stopped consumer");

kafka/src/kafka/impl/consumer_impl.cpp

@@ -5,16 +5,16 @@
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-#include <boost/range/iterator_range.hpp>
-
 #include <userver/kafka/impl/configuration.hpp>
+#include <userver/kafka/impl/offset.hpp>
 #include <userver/kafka/impl/stats.hpp>
 #include <userver/logging/log.hpp>
 #include <userver/testsuite/testpoint.hpp>
 #include <userver/tracing/span.hpp>
 #include <userver/utils/fast_scope_guard.hpp>
 #include <userver/utils/span.hpp>
 
+#include <kafka/impl/error_buffer.hpp>
 #include <kafka/impl/holders_aliases.hpp>
 #include <kafka/impl/log_level.hpp>
 
@@ -294,51 +294,57 @@ void ConsumerImpl::Commit() { rd_kafka_commit(consumer_.GetHandle(), nullptr, /*
 
 void ConsumerImpl::AsyncCommit() { rd_kafka_commit(consumer_.GetHandle(), nullptr, /*async=*/1); }
 
-std::pair<std::int64_t, std::int64_t> ConsumerImpl::GetMinMaxOffset(const std::string& topic, std::int32_t partition) {
-    std::int64_t offset_high{-1};
-    std::int64_t offset_low{-1};
-
+OffsetRange ConsumerImpl::GetOffsetRange(const std::string& topic, std::int32_t partition) const {
+    OffsetRange offset_range;
     auto err = rd_kafka_query_watermark_offsets(
-        consumer_.GetHandle(), topic.data(), partition, &offset_low, &offset_high, /*timeout_ms=*/-1
+        consumer_.GetHandle(),
+        topic.с_str(),
+        partition,
+        &offset_range.low,
+        &offset_range.high,
+        /*timeout_ms=*/-1
     );
     if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
-        LOG_ERROR() << fmt::format("Failed to get offsets: {}", rd_kafka_err2str(err));
-        return {-1, -1};
+        PrintErrorAndThrow("get offsets", rd_kafka_err2str(err));
     }
 
-    if (offset_high == RD_KAFKA_OFFSET_INVALID) {
-        LOG_ERROR() << fmt::format("Invalid offset for topic {} partition {}", topic, partition);
-        return {-1, -1};
+    if (offset_range.low == RD_KAFKA_OFFSET_INVALID || offset_range.high == RD_KAFKA_OFFSET_INVALID) {
+        PrintErrorAndThrow("get offsets", fmt::format("invalid offset for topic {} partition {}", topic, partition));
     }
 
-    return {offset_low, offset_high};
+    return offset_range;
 }
 
-std::vector<std::int32_t> ConsumerImpl::GetPartitionsId(const std::string& topic) {
-    const rd_kafka_metadata_t* raw_metadata{nullptr};
-    auto err = rd_kafka_metadata(consumer_.GetHandle(), 0, nullptr, &raw_metadata, /*timeout_ms=*/-1);
-    if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
-        LOG_ERROR() << fmt::format("Failed to fetch metadata: ", rd_kafka_err2str(err));
-        return {};
-    }
-    MetadataHolder metadata{raw_metadata};
-
-    auto topics = boost::make_iterator_range(metadata->topics, metadata->topics + metadata->topic_cnt);
-    auto* topic_it = std::find_if(topics.begin(), topics.end(), [&topic](const rd_kafka_metadata_topic& topic_raw) {
-        return topic == topic_raw.topic;
-    });
+std::vector<std::uint32_t> ConsumerImpl::GetPartitionIds(const std::string& topic) const {
+    MetadataHolder metadata{[&consumer_] {
+        const rd_kafka_metadata_t* raw_metadata{nullptr};
+        auto err = rd_kafka_metadata(consumer_.GetHandle(), 0, nullptr, &raw_metadata, /*timeout_ms=*/-1);
+        if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
+            PrintErrorAndThrow("fetch metadata", rd_kafka_err2str(err));
+        }
+        return raw_metadata;
+    }()};
+
+    userver::utils::span<const rd_kafka_metadata_topic> topics{
+        metadata->topics, static_cast<std::size_t>(metadata->topic_cnt)};
+    const auto* topic_it =
+        std::find_if(topics.begin(), topics.end(), [&topic](const rd_kafka_metadata_topic& topic_raw) {
+            return topic == topic_raw.topic;
+        });
     if (topic_it == topics.end()) {
-        LOG_ERROR() << fmt::format("Topic not found: {}", topic);
-        return {};
+        PrintErrorAndThrow("find topic", fmt::format("{} not found", topic));
     }
 
-    std::vector<std::int32_t> partitions_id;
-    for (const auto& partition :
-         boost::make_iterator_range(topic_it->partitions, topic_it->partitions + topic_it->partition_cnt)) {
-        partitions_id.push_back(partition.id);
+    userver::utils::span<const rd_kafka_metadata_partition> partitions{
+        topic_it->partitions, static_cast<std::size_t>(topic_it->partition_cnt)};
+    std::vector<std::uint32_t> partition_ids;
+    partition_ids.reserve(partitions.size());
+
+    for (const auto& partition : partitions) {
+        partition_ids.push_back(partition.id);
     }
 
-    return partitions_id;
+    return partition_ids;
 }
 
 EventHolder ConsumerImpl::PollEvent() {

kafka/src/kafka/impl/consumer_impl.hpp

@@ -18,6 +18,7 @@ namespace kafka::impl {
 
 struct Stats;
 struct TopicStats;
+struct OffsetRange;
 
 /// @brief Consumer implementation based on `librdkafka`.
 /// @warning All methods calls the `librdkafka` functions that very often uses
@@ -37,11 +38,11 @@ class ConsumerImpl final {
     /// @brief Schedules the commitment task.
     void AsyncCommit();
 
-    /// @brief Retrieves the minimum and maximum offsets for the specified Kafka topic and partition.
-    std::pair<std::int64_t, std::int64_t> GetMinMaxOffset(const std::string& topic, std::int32_t partition);
+    /// @brief Retrieves the low and high offsets for the specified kafka topic and partition.
+    OffsetRange GetOffsetRange(const std::string& topic, std::int32_t partition) const;
 
-    /// @brief Retrieves the partition IDs for the specified Kafka topic.
-    std::vector<std::int32_t> GetPartitionsId(const std::string& topic);
+    /// @brief Retrieves the partition IDs for the specified kafka topic.
+    std::vector<std::uint32_t> GetPartitionIds(const std::string& topic) const;
 
     /// @brief Effectively calls `PollMessage` until `deadline` is reached
     /// and no more than `max_batch_size` messages polled.

kafka/src/kafka/impl/error_buffer.cpp

@@ -18,6 +18,12 @@ void PrintErrorAndThrow(std::string_view failed_action, const ErrorBuffer& err_b
     throw std::runtime_error{full_error};
 }
 
+void PrintErrorAndThrow(std::string_view failed_action, std::string_view reason) {
+    const auto full_error = fmt::format("Failed to {}: {}", failed_action, reason);
+    LOG_ERROR() << full_error;
+    throw std::runtime_error{full_error};
+}
+
 }  // namespace kafka::impl
 
 USERVER_NAMESPACE_END

kafka/src/kafka/impl/error_buffer.hpp

@@ -17,6 +17,7 @@ constexpr std::size_t kErrorBufferSize = 512;
 using ErrorBuffer = std::array<char, kErrorBufferSize>;
 
 [[noreturn]] void PrintErrorAndThrow(std::string_view failed_action, const ErrorBuffer& err_buf);
+[[noreturn]] void PrintErrorAndThrow(std::string_view failed_action, std::string_view reason);
 
 }  // namespace kafka::impl