feat kafka: add bulk send producer method

Добавлен метод `Producer::Send` для синхронной отправки пачки сообщений
в указанный топик Kafka. В случае ошибки отправки выкидывается исключение
`BulkSendException`, которое содержит подробности об ошибках отправки
сообщений из пачки.

`Producer::Send` пытается обработать ошибки переполнения локальной
очереди librdkafka в пределах настроенного `delivery.timeout.ms`.

В отличие от старых методов `Send` и `SendAsync` метод порождает
только одну корутину на всю пачку сообщений.

Author: disaykin

kafka/include/userver/kafka/exceptions.hpp

@@ -1,6 +1,8 @@
 #pragma once
 
 #include <cstdint>
+#include <exception>
+#include <map>
 #include <stdexcept>
 #include <string_view>
 
@@ -27,6 +29,29 @@ class SendException : public std::runtime_error {
     const bool is_retryable_{false};
 };
 
+/// @brief Base exception thrown by Producer::Send in bulk mode
+/// in case of one or more send errors.
+class BulkSendException : public std::runtime_error {
+    static constexpr const char* kWhat{"Some messages was not delivered."};
+
+public:
+    using ExceptionMap = std::map<std::size_t, std::exception_ptr>;
+
+    BulkSendException(ExceptionMap nested_exceptions);
+
+    /// @return nested errors.
+    /// Nested exceptions are subclasses of SendException.
+    const ExceptionMap& GetExceptions() const noexcept;
+
+private:
+    /// @brief A mapping from the message's index in the bulk send operation
+    /// to the exception that occurred during its delivering.
+    /// @details Key: 0-based index of the element in the input batch.
+    /// Value: Pointer to the exception.
+    /// @note Contains only indices that resulted in an error.
+    const ExceptionMap nested_exceptions_;
+};
+
 class DeliveryTimeoutException final : public SendException {
     static constexpr const char* kWhat{
         "Message is not delivered after `delivery_timeout` milliseconds. Hint: "

kafka/include/userver/kafka/producer.hpp

@@ -7,6 +7,7 @@
 #include <userver/kafka/exceptions.hpp>
 #include <userver/kafka/headers.hpp>
 #include <userver/utils/fast_pimpl.hpp>
+#include <userver/utils/span.hpp>
 #include <userver/utils/statistics/writer.hpp>
 
 USERVER_NAMESPACE_BEGIN
@@ -110,6 +111,36 @@ class Producer final {
         HeaderViews headers = {}
     ) const;
 
+    /// @brief Sends given messages to topic `topic_name` by given `key`
+    /// and `partition` (if passed) with payload contains the `messages`
+    /// data. Asynchronously waits until the messages are delivered or the delivery
+    /// error occurred.
+    ///
+    /// No payload data is copied. Method holds the data until messages are
+    /// delivered.
+    ///
+    /// Thread-safe and can be called from any number of threads
+    /// concurrently.
+    ///
+    /// If `partition` not passed, partition is chosen by internal
+    /// Kafka partitioner.
+    ///
+    /// @warning if `enable_idempotence` option is enabled, do not use both
+    /// explicit partitions and Kafka-chosen ones.
+    ///
+    /// @throws BulkSendException if some messages was not delivered
+    /// and acked by Kafka Broker in configured timeout.
+    ///
+    /// @note Use BulkSendException::GetExceptions method to get a list
+    /// of occured nested exceptions.
+    void Send(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        utils::span<const std::string> messages,
+        std::optional<std::uint32_t> partition = kUnassignedPartition,
+        HeaderViews headers = {}
+    ) const;
+
     /// @brief Same as Producer::Send, but returns the task which can be
     /// used to wait the message delivery manually.
     ///
@@ -141,6 +172,14 @@ class Producer final {
         impl::HeadersHolder&& headers_holder
     ) const;
 
+    void SendImpl(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        utils::span<const std::string> messages,
+        std::optional<std::uint32_t> partition,
+        std::vector<impl::HeadersHolder>&& headers_holders
+    ) const;
+
     const std::string name_;
     engine::TaskProcessor& producer_task_processor_;
 

kafka/src/kafka/exceptions.cpp

@@ -13,6 +13,13 @@ SendException::SendException(const char* what, bool is_retryable)
       is_retryable_(is_retryable)
 {}
 
+BulkSendException::BulkSendException(BulkSendException::ExceptionMap nested_exceptions)
+    : std::runtime_error(kWhat),
+      nested_exceptions_(std::move(nested_exceptions))
+{}
+
+const BulkSendException::ExceptionMap& BulkSendException::GetExceptions() const noexcept { return nested_exceptions_; }
+
 DeliveryTimeoutException::DeliveryTimeoutException()
     : SendException(kWhat, /*is_retryable=*/true)
 {}

kafka/src/kafka/impl/producer_impl.cpp

@@ -8,6 +8,7 @@
 #include <userver/engine/wait_any.hpp>
 #include <userver/kafka/impl/configuration.hpp>
 #include <userver/tracing/span.hpp>
+#include <userver/utils/span.hpp>
 #include <userver/utils/trivial_map.hpp>
 
 #include <kafka/impl/log_level.hpp>
@@ -138,20 +139,56 @@ DeliveryResult ProducerImpl::Send(
     HeadersHolder headers_holder
 ) const {
     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));
+        ScheduleMessageDelivery(topic_name, key, message, partition, std::move(headers_holder), deadline);
 
     WaitUntilDeliveryReported(delivery_result_future);
 
     return delivery_result_future.get();
 }
 
+std::vector<DeliveryResult> ProducerImpl::Send(
+    utils::zstring_view topic_name,
+    std::string_view key,
+    utils::span<const std::string> messages,
+    std::optional<std::uint32_t> partition,
+    std::vector<HeadersHolder> headers_holders
+) const {
+    UASSERT(messages.size() == headers_holders.size());
+
+    LOG(operation_log_level_) <<
+        fmt::format("Messages {} to topic '{}' are requested to send", messages.size(), topic_name);
+
+    std::vector<engine::Future<DeliveryResult>> delivery_result_futures;
+    delivery_result_futures.reserve(messages.size());
+
+    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)
+        );
+    }
+
+    std::vector<DeliveryResult> delivery_results;
+    delivery_results.reserve(messages.size());
+
+    for (auto& delivery_result_future : delivery_result_futures) {
+        WaitUntilDeliveryReported(delivery_result_future);
+
+        delivery_results.emplace_back(delivery_result_future.get());
+    }
+
+    return delivery_results;
+}
+
 engine::Future<DeliveryResult> ProducerImpl::ScheduleMessageDelivery(
     utils::zstring_view topic_name,
     std::string_view key,
     std::string_view message,
     std::optional<std::uint32_t> partition,
-    HeadersHolder headers_holder
+    HeadersHolder headers_holder,
+    engine::Deadline deadline
 ) const {
     auto waiter = std::make_unique<DeliveryWaiter>();
     auto wait_handle = waiter->GetFuture();
@@ -181,6 +218,7 @@ engine::Future<DeliveryResult> ProducerImpl::ScheduleMessageDelivery(
     ///
     /// Headers holder **must** be released if `rd_kafka_producev` succeeded.
 
+    while (!deadline.IsReached() && !engine::current_task::ShouldCancel()) {
 #ifdef __clang__
 #pragma clang diagnostic push
 #pragma clang diagnostic ignored "-Wgnu-statement-expression"
@@ -203,12 +241,19 @@ engine::Future<DeliveryResult> ProducerImpl::ScheduleMessageDelivery(
 #pragma clang diagnostic pop
 #endif
 
-    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 {
-        LOG_WARNING("Failed to enqueue message to Kafka local queue: {}", rd_kafka_err2str(enqueue_error));
-        waiter->SetDeliveryResult(DeliveryResult{enqueue_error});
+        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 (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::InterruptibleSleepFor(std::chrono::milliseconds{10});
+            continue;
+        } else {
+            LOG_WARNING("Failed to enqueue message to Kafka local queue: {}", rd_kafka_err2str(enqueue_error));
+            waiter->SetDeliveryResult(DeliveryResult{enqueue_error});
+        }
+        break;
     }
 
     return wait_handle;

kafka/src/kafka/impl/producer_impl.hpp

@@ -6,9 +6,11 @@
 
 #include <librdkafka/rdkafka.h>
 
+#include <userver/engine/deadline.hpp>
 #include <userver/kafka/impl/stats.hpp>
 #include <userver/logging/level.hpp>
 #include <userver/utils/periodic_task.hpp>
+#include <userver/utils/span.hpp>
 
 #include <kafka/impl/concurrent_event_waiter.hpp>
 #include <kafka/impl/delivery_waiter.hpp>
@@ -40,6 +42,16 @@ class ProducerImpl final {
         HeadersHolder headers
     ) const;
 
+    /// @brief Send messages and waits for its delivery.
+    /// While waiting handles other messages delivery reports, errors and logs.
+    [[nodiscard]] std::vector<DeliveryResult> Send(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        utils::span<const std::string> messages,
+        std::optional<std::uint32_t> partition,
+        std::vector<HeadersHolder> headers
+    ) const;
+
     /// @brief Waits until scheduled messages are delivered for
     /// at most 2 x `delivery_timeout`.
     ///
@@ -58,7 +70,8 @@ class ProducerImpl final {
         std::string_view key,
         std::string_view message,
         std::optional<std::uint32_t> partition,
-        HeadersHolder headers
+        HeadersHolder headers,
+        engine::Deadline deadline
     ) const;
 
     /// @brief Poll a delivery or error event from producer's queue.

kafka/src/kafka/producer.cpp

@@ -7,6 +7,7 @@
 #include <userver/testsuite/testpoint.hpp>
 #include <userver/tracing/span.hpp>
 #include <userver/utils/async.hpp>
+#include <userver/utils/span.hpp>
 #include <userver/utils/text_light.hpp>
 
 #include <kafka/impl/producer_impl.hpp>
@@ -54,26 +55,52 @@ void SendToTestPoint(
     }());
 }
 
-[[noreturn]] void ThrowSendError(const impl::DeliveryResult& delivery_result) {
+std::exception_ptr BuildSendError(const impl::DeliveryResult& delivery_result) {
     const auto error = delivery_result.GetMessageError();
     UASSERT(error != RD_KAFKA_RESP_ERR_NO_ERROR);
 
     switch (error) {
         case RD_KAFKA_RESP_ERR__MSG_TIMED_OUT:
-            throw DeliveryTimeoutException{};
+            return std::make_exception_ptr(DeliveryTimeoutException{});
         case RD_KAFKA_RESP_ERR__QUEUE_FULL:
-            throw QueueFullException{};
+            return std::make_exception_ptr(QueueFullException{});
         case RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE:
-            throw MessageTooLargeException{};
+            return std::make_exception_ptr(MessageTooLargeException{});
         case RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC:
-            throw kafka::UnknownTopicException{};
+            return std::make_exception_ptr(UnknownTopicException{});
         case RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION:
-            throw kafka::UnknownPartitionException{};
+            return std::make_exception_ptr(UnknownPartitionException{});
         default:
-            throw kafka::SendException{rd_kafka_err2str(error)};
+            return std::make_exception_ptr(SendException{rd_kafka_err2str(error)});
+    }
+}
+
+[[noreturn]] void ThrowSendError(const impl::DeliveryResult& delivery_result) {
+    std::rethrow_exception(BuildSendError(delivery_result));
+}
+
+auto BuildOwningHeaders(const impl::HeadersHolder& headers) {
+    auto reader = HeadersReader{headers.GetHandle()};
+    return std::vector<OwningHeader>{reader.begin(), reader.end()};
+}
+
+auto BuildHeaderHolders(HeaderViews headers, std::size_t count) {
+    std::vector<impl::HeadersHolder> holders;
+    holders.reserve(count);
+    for (std::size_t i = 0; i < count; ++i) holders.emplace_back(impl::HeadersHolder{headers});
+    return holders;
+}
+
+void HandleDeliveryErrors(const std::vector<impl::DeliveryResult>& delivery_results) {
+    std::map<std::size_t, std::exception_ptr> exceptions;
+    for (std::size_t i = 0; i < delivery_results.size(); ++i) {
+        const auto& delivery_result = delivery_results[i];
+        if (!delivery_result.IsSuccess()) {
+            exceptions[i] = BuildSendError(delivery_result);
+        }
     }
 
-    UASSERT(false);
+    if (!exceptions.empty()) throw BulkSendException{std::move(exceptions)};
 }
 
 }  // namespace
@@ -111,6 +138,22 @@ void Producer::Send(
     }).Get();
 }
 
+void Producer::Send(
+    utils::zstring_view topic_name,
+    std::string_view key,
+    utils::span<const std::string> messages,
+    std::optional<std::uint32_t> partition,
+    HeaderViews headers
+) const {
+    utils::Async(
+        producer_task_processor_,
+        "producer_send_bulk",
+        [this, topic_name, key, messages, partition, &headers] {
+            SendImpl(topic_name, key, messages, partition, BuildHeaderHolders(headers, messages.size()));
+        }
+    ).Get();
+}
+
 engine::TaskWithResult<void> Producer::SendAsync(
     std::string topic_name,
     std::string key,
@@ -148,8 +191,7 @@ void Producer::SendImpl(
 
     std::vector<OwningHeader> headers_copy;
     if (testsuite::AreTestpointsAvailable()) {
-        auto reader = HeadersReader{headers_holder.GetHandle()};
-        headers_copy = std::vector<OwningHeader>{reader.begin(), reader.end()};
+        headers_copy = BuildOwningHeaders(headers_holder);
     }
     if (testsuite::AreTestpointsAvailable()) {
         SendToTestPoint(name_, topic_name, key, message, partition, headers_copy, "::started");
@@ -166,6 +208,40 @@ void Producer::SendImpl(
     }
 }
 
+void Producer::SendImpl(
+    utils::zstring_view topic_name,
+    std::string_view key,
+    utils::span<const std::string> messages,
+    std::optional<std::uint32_t> partition,
+    std::vector<impl::HeadersHolder>&& headers_holders
+) const {
+    tracing::Span::CurrentSpan().AddTag("kafka_producer", name_);
+    tracing::Span::CurrentSpan().AddTag("kafka_send_key", std::string{key});
+
+    std::vector<std::vector<OwningHeader>> headers_copies;
+    if (testsuite::AreTestpointsAvailable()) {
+        for (std::size_t i = 0; i < messages.size(); ++i) {
+            headers_copies.emplace_back(BuildOwningHeaders(headers_holders[i]));
+        }
+    }
+    if (testsuite::AreTestpointsAvailable()) {
+        for (std::size_t i = 0; i < messages.size(); ++i) {
+            SendToTestPoint(name_, topic_name, key, messages[i], partition, headers_copies[i], "::started");
+        }
+    }
+
+    const std::vector<impl::DeliveryResult> delivery_results =
+        producer_->Send(topic_name, key, messages, partition, std::move(headers_holders));
+
+    HandleDeliveryErrors(delivery_results);
+
+    if (testsuite::AreTestpointsAvailable()) {
+        for (std::size_t i = 0; i < messages.size(); ++i) {
+            SendToTestPoint(name_, topic_name, key, messages[i], partition, headers_copies[i]);
+        }
+    }
+}
+
 }  // namespace kafka
 
 USERVER_NAMESPACE_END