WiP

Author: disaykin

kafka/include/userver/kafka/producer.hpp

@@ -104,7 +104,22 @@ class Producer final {
     ///
     /// @note Use SendException::IsRetryable method to understand whether there is
     /// a sense to retry the message sending.
-    /// @snippet kafka/tests/producer_kafkatest.cpp Producer retryable error
+    ///
+    /// @code{.cpp}
+    /// bool delivered{false};
+    /// while (!delivered && !deadline.IsReached()) {
+    ///     try {
+    ///         producer.Send(topic, key, message);
+    ///         delivered = true;
+    ///     } catch (const kafka::SendException& e) {
+    ///         if (e.IsRetryable()) {
+    ///             engine::InterruptibleSleepFor(std::chrono::milliseconds{10});
+    ///             continue;
+    ///         }
+    ///         break;
+    ///     }
+    /// }
+    /// @endcode
     void Send(
         utils::zstring_view topic_name,
         std::string_view key,

kafka/src/kafka/impl/producer_impl.cpp

@@ -140,7 +140,7 @@ DeliveryResult ProducerImpl::Send(
     LOG(operation_log_level_) << fmt::format("Message to topic '{}' is requested to send", topic_name);
     auto deadline = engine::Deadline::FromDuration(delivery_timeout_);
     auto delivery_result_future =
-        ScheduleMessageDelivery(topic_name, key, message, partition, std::move(headers_holder), deadline, QueueFullHandlingPolicy::Throw);
+        ScheduleMessageDelivery(topic_name, key, message, partition, std::move(headers_holder), deadline);
 
     WaitUntilDeliveryReported(delivery_result_future);
 
@@ -165,7 +165,7 @@ std::vector<DeliveryResult> ProducerImpl::Send(
     auto deadline = engine::Deadline::FromDuration(delivery_timeout_);
     for (std::size_t i = 0; i < messages.Size(); ++i) {
         delivery_result_futures.emplace_back(
-            ScheduleMessageDelivery(topic_name, key, messages[i], partition, std::move(headers_holders[i]), deadline, QueueFullHandlingPolicy::Retry)
+            ScheduleMessageDelivery(topic_name, key, messages[i], partition, std::move(headers_holders[i]), deadline)
         );
     }
 
@@ -187,8 +187,7 @@ engine::Future<DeliveryResult> ProducerImpl::ScheduleMessageDelivery(
     std::string_view message,
     std::optional<std::uint32_t> partition,
     HeadersHolder headers_holder,
-    engine::Deadline deadline,
-    QueueFullHandlingPolicy queue_full_handling_policy
+    engine::Deadline deadline
 ) const {
     auto waiter = std::make_unique<DeliveryWaiter>();
     auto wait_handle = waiter->GetFuture();
@@ -244,7 +243,7 @@ engine::Future<DeliveryResult> ProducerImpl::ScheduleMessageDelivery(
         if (enqueue_error == RD_KAFKA_RESP_ERR_NO_ERROR) {
             [[maybe_unused]] const auto headers_holder_ptr = headers_holder.release();
             [[maybe_unused]] const auto waiter_ptr = waiter.release();
-        } else if (queue_full_handling_policy == QueueFullHandlingPolicy::Retry && enqueue_error == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
+        } else if (enqueue_error == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
             LOG_LIMITED_WARNING("Kafka local queue is full");
             /// waiting for a while for the queue to clear up
             engine::Yield();

kafka/src/kafka/impl/producer_impl.hpp

@@ -63,11 +63,6 @@ class ProducerImpl final {
     void EventCallback();
 
 private:
-    enum class QueueFullHandlingPolicy {
-        Throw,
-        Retry,
-    };
-
     /// @brief Schedules the message delivery.
     /// @returns the future for delivery result, which must be awaited.
     [[nodiscard]] engine::Future<DeliveryResult> ScheduleMessageDelivery(
@@ -76,8 +71,7 @@ class ProducerImpl final {
         std::string_view message,
         std::optional<std::uint32_t> partition,
         HeadersHolder headers,
-        engine::Deadline deadline,
-        QueueFullHandlingPolicy queue_full_handling_policy
+        engine::Deadline deadline
     ) const;
 
     /// @brief Poll a delivery or error event from producer's queue.

kafka/tests/producer_kafkatest.cpp

@@ -200,32 +200,9 @@ UTEST_F(ProducerTest, FullQueue) {
             .emplace_back(producer.SendAsync(topic, fmt::format("test-key-{}", send), fmt::format("test-msg-{}", send))
             );
     }
-    auto make_send_request =
-        [&producer,
-         &topic,
-         key = fmt::format("test-key-{}", kMaxQueueMessages),
-         message = fmt::format("test-msg-{}", kMaxQueueMessages)] { producer.Send(topic, key, message); };
-
-    UEXPECT_THROW(make_send_request(), kafka::QueueFullException);
-
-    /// [Producer retryable error]
-    bool delivered{false};
-    const auto deadline = engine::Deadline::FromDuration(producer_configuration.delivery_timeout);
-    while (!delivered && !deadline.IsReached()) {
-        try {
-            make_send_request();
-            delivered = true;
-        } catch (const kafka::SendException& e) {
-            if (e.IsRetryable()) {
-                engine::InterruptibleSleepFor(std::chrono::milliseconds{10});
-                continue;
-            }
-            break;
-        }
-    }
-    /// [Producer retryable error]
 
-    EXPECT_TRUE(delivered);
+    UEXPECT_NO_THROW(producer.Send(
+        topic, fmt::format("test-key-{}", kMaxQueueMessages), fmt::format("test-msg-{}", kMaxQueueMessages)));
     UEXPECT_NO_THROW(engine::WaitAllChecked(results));
 }