Fix comments

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

kafka/include/userver/kafka/consumer_scope.hpp

@@ -116,7 +116,7 @@ class ConsumerScope final {
     void AsyncCommit();
 
     /// @brief Retrieves the minimum and maximum offsets for the specified topic and partition.
-    /// @throws GetOffsetRangeException if the offsets could not retrieve or if the returned offsets are invalid
+    /// @throws GetOffsetRangeException if the offsets could not be retrieve or if the returned offsets are invalid
     /// @warning This is a blocking call
     /// @param topic The name of the topic.
     /// @param partition The partition number of the topic.

kafka/include/userver/kafka/exceptions.hpp

@@ -1,6 +1,8 @@
 #pragma once
 
+#include <cstdint>
 #include <stdexcept>
+#include <string_view>
 
 USERVER_NAMESPACE_BEGIN
 
@@ -68,25 +70,41 @@ class UnknownPartitionException final : public SendException {
     UnknownPartitionException();
 };
 
-class GetOffsetRangeException : public std::runtime_error {
+/// @brief Exception thrown when there is an error retrieving the offset range.
+class OffsetRangeException : public std::runtime_error {
 public:
     using std::runtime_error::runtime_error;
 
-    GetOffsetRangeException(const char* what) : std::runtime_error(what) {}
+public:
+    OffsetRangeException(std::string_view what, std::string_view topic, std::uint32_t partition);
+};
+
+class OffsetRangeTimeoutException final : public OffsetRangeException {
+    static constexpr const char* kWhat = "Timeout while fetching offsets.";
+
+public:
+    OffsetRangeTimeoutException(std::string_view topic, std::uint32_t partition);
 };
 
 class TopicNotFoundException final : public std::runtime_error {
 public:
     using std::runtime_error::runtime_error;
-
-    TopicNotFoundException(const char* what) : std::runtime_error(what) {}
 };
 
+/// @brief Exception thrown when fetching metadata.
 class GetMetadataException final : public std::runtime_error {
 public:
     using std::runtime_error::runtime_error;
 
-    GetMetadataException(const char* what) : std::runtime_error(what) {}
+public:
+    GetMetadataException(std::string_view what, std::string_view topic);
+};
+
+class GetMetadataTimeoutException final : public std::runtime_error {
+    static constexpr const char* kWhat = "Timeout while getting metadata.";
+
+public:
+    GetMetadataTimeoutException(std::string_view topic);
 };
 
 }  // namespace kafka

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

@@ -55,14 +55,14 @@ class Consumer final {
     /// @param topics stands for topics list that consumer subscribes to
     /// after ConsumerScope::Start called
     /// @param consumer_task_processor -- task processor for message batches
-    /// @param consumer_operation_task_processor -- task processor for consumer operation
+    /// @param consumer_blocking_task_processor -- task processor for consumer blocking operation
     /// polling
     /// All callbacks are invoked in `main_task_processor`
     Consumer(
         const std::string& name,
         const std::vector<std::string>& topics,
         engine::TaskProcessor& consumer_task_processor,
-        engine::TaskProcessor& consumer_operation_task_processor,
+        engine::TaskProcessor& consumer_blocking_task_processor,
         engine::TaskProcessor& main_task_processor,
         const ConsumerConfiguration& consumer_configuration,
         const Secret& secrets,

kafka/include/userver/kafka/offset_range.hpp

@@ -6,6 +6,7 @@ USERVER_NAMESPACE_BEGIN
 
 namespace kafka {
 
+/// @brief Represents the range of offsets.
 struct OffsetRange final {
     ///@brief The low watermark offset. It indicates the earliest available offset in Kafka.
     std::uint32_t low{};

kafka/src/kafka/consumer_component.cpp

@@ -22,7 +22,7 @@ ConsumerComponent::ConsumerComponent(
           config.Name(),
           config["topics"].As<std::vector<std::string>>(),
           context.GetTaskProcessor("consumer-task-processor"),
-          context.GetTaskProcessor("consumer-operation-task-processor"),
+          context.GetTaskProcessor("consumer-blocking-task-processor"),
           context.GetTaskProcessor("main-task-processor"),
           config.As<impl::ConsumerConfiguration>(),
           context.FindComponent<components::Secdist>().Get().Get<impl::BrokerSecrets>().GetSecretByComponentName(

kafka/src/kafka/exceptions.cpp

@@ -1,5 +1,7 @@
 #include <userver/kafka/exceptions.hpp>
 
+#include <fmt/format.h>
+
 USERVER_NAMESPACE_BEGIN
 
 namespace kafka {
@@ -19,6 +21,18 @@ UnknownTopicException::UnknownTopicException() : SendException(kWhat) {}
 
 UnknownPartitionException::UnknownPartitionException() : SendException(kWhat) {}
 
+OffsetRangeException::OffsetRangeException(std::string_view what, std::string_view topic, std::uint32_t partition)
+    : std::runtime_error(fmt::format("{} topic: '{}', partition: {}", what, topic, partition)) {}
+
+OffsetRangeTimeoutException::OffsetRangeTimeoutException(std::string_view topic, std::uint32_t partition)
+    : OffsetRangeException(kWhat, topic, partition) {}
+
+GetMetadataException::GetMetadataException(std::string_view what, std::string_view topic)
+    : std::runtime_error(fmt::format("{} topic: '{}'", what, topic)) {}
+
+GetMetadataTimeoutException::GetMetadataTimeoutException(std::string_view topic)
+    : std::runtime_error(fmt::format("{} topic: '{}'", kWhat, topic)) {}
+
 }  // namespace kafka
 
 USERVER_NAMESPACE_END

kafka/src/kafka/impl/consumer.cpp

@@ -53,7 +53,7 @@ Consumer::Consumer(
     const std::string& name,
     const std::vector<std::string>& topics,
     engine::TaskProcessor& consumer_task_processor,
-    engine::TaskProcessor& consumer_operation_task_processor,
+    engine::TaskProcessor& consumer_blocking_task_processor,
     engine::TaskProcessor& main_task_processor,
     const ConsumerConfiguration& configuration,
     const Secret& secrets,
@@ -63,7 +63,7 @@ Consumer::Consumer(
       topics_(topics),
       execution_params(params),
       consumer_task_processor_(consumer_task_processor),
-      consumer_operation_task_processor_(consumer_operation_task_processor),
+      consumer_operation_task_processor_(consumer_blocking_task_processor),
       main_task_processor_(main_task_processor),
       conf_(Configuration{name, configuration, secrets}.Release()) {
     /// To check configuration validity

kafka/src/kafka/impl/consumer_impl.cpp

@@ -300,8 +300,8 @@ OffsetRange ConsumerImpl::GetOffsetRange(
     std::uint32_t partition,
     std::optional<std::chrono::milliseconds> timeout
 ) const {
-    int64_t low_offset{0};
-    int64_t high_offset{0};
+    std::int64_t low_offset{0};
+    std::int64_t high_offset{0};
 
     auto err = rd_kafka_query_watermark_offsets(
         consumer_.GetHandle(),
@@ -311,14 +311,15 @@ OffsetRange ConsumerImpl::GetOffsetRange(
         &high_offset,
         static_cast<int>(timeout.value_or(std::chrono::milliseconds(-1)).count())
     );
+
+    if (err == RD_KAFKA_RESP_ERR__TIMED_OUT) {
+        throw OffsetRangeTimeoutException{topic, partition};
+    }
     if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
-        throw GetOffsetRangeException{fmt::format("Failed to get offsets: {}", rd_kafka_err2str(err))};
+        throw OffsetRangeException{fmt::format("Failed to get offsets: {}", rd_kafka_err2str(err)), topic, partition};
     }
-
     if (low_offset == RD_KAFKA_OFFSET_INVALID || high_offset == RD_KAFKA_OFFSET_INVALID) {
-        throw GetOffsetRangeException{
-            fmt::format("Failed to get offsets: invalid offset for topic {} partition {}", topic, partition)
-        };
+        throw OffsetRangeException{fmt::format("Failed to get offsets: invalid offset."), topic, partition};
     }
 
     return {static_cast<std::uint32_t>(low_offset), static_cast<std::uint32_t>(high_offset)};
@@ -336,8 +337,11 @@ ConsumerImpl::GetPartitionIds(const std::string& topic, std::optional<std::chron
             &raw_metadata,
             static_cast<int>(timeout.value_or(std::chrono::milliseconds(-1)).count())
         );
+        if (err == RD_KAFKA_RESP_ERR__TIMED_OUT) {
+            throw GetMetadataTimeoutException{topic};
+        }
         if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
-            throw GetMetadataException{fmt::format("Failed to fetch metadata: {}", rd_kafka_err2str(err))};
+            throw GetMetadataException{fmt::format("Failed to fetch metadata: {}.", rd_kafka_err2str(err)), topic};
         }
         return raw_metadata;
     }()};