feat kafka: add bulk send producer method

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

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

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

Tests: протестировано CI

---

Pull Request resolved: #1134

Co-authored-by: antoshkka <antoshkka@userver.tech>
Co-authored-by: antoshkka <antoshkka@userver.tech>
Co-authored-by: antoshkka <antoshkka@userver.tech>
commit_hash:8c87611375eb1d0a8a8f7a67731e48d91f023beb

Author: 2 people

.mapping.json

@@ -2778,6 +2778,7 @@
   "kafka/include/userver/kafka/impl/consumer.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/impl/consumer.hpp",
   "kafka/include/userver/kafka/impl/consumer_params.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/impl/consumer_params.hpp",
   "kafka/include/userver/kafka/impl/holders.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/impl/holders.hpp",
+  "kafka/include/userver/kafka/impl/messages.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/impl/messages.hpp",
   "kafka/include/userver/kafka/impl/stats.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/impl/stats.hpp",
   "kafka/include/userver/kafka/message.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/message.hpp",
   "kafka/include/userver/kafka/offset_range.hpp":"taxi/uservices/userver/kafka/include/userver/kafka/offset_range.hpp",

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>;
+
+    explicit 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/impl/messages.hpp

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <cstdint>
+#include <string_view>
+#include <type_traits>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace kafka::impl {
+
+/// @brief Message collection adapter.
+class Messages {
+public:
+    virtual std::size_t Size() const noexcept = 0;
+    virtual std::string_view operator[](std::size_t index) const noexcept = 0;
+
+protected:
+    ~Messages() = default;
+};
+
+template <typename Container>
+class MessagesAdapter final : public Messages {
+public:
+    static_assert(
+        std::is_convertible_v<decltype(std::declval<const Container&>()[0]), std::string_view>,
+        "Container must support operator[] and return a type convertible to std::string_view"
+    );
+
+    static_assert(
+        std::is_integral_v<decltype(std::declval<const Container&>().size())>,
+        "Container must support method size() and return an integral type"
+    );
+
+    explicit MessagesAdapter(const Container& data) noexcept : data_{data} {}
+
+    std::size_t Size() const noexcept override { return data_.size(); }
+    std::string_view operator[](std::size_t index) const noexcept override { return data_[index]; }
+
+private:
+    const Container& data_;
+};
+
+}  // namespace kafka::impl
+
+USERVER_NAMESPACE_END

kafka/include/userver/kafka/producer.hpp

@@ -1,11 +1,14 @@
 #pragma once
 
 #include <cstdint>
+#include <type_traits>
+#include <utility>
 
 #include <userver/engine/task/task_processor_fwd.hpp>
 #include <userver/engine/task/task_with_result.hpp>
 #include <userver/kafka/exceptions.hpp>
 #include <userver/kafka/headers.hpp>
+#include <userver/kafka/impl/messages.hpp>
 #include <userver/utils/fast_pimpl.hpp>
 #include <userver/utils/statistics/writer.hpp>
 
@@ -81,26 +84,20 @@ class Producer final {
 
     /// @brief Sends given message to topic `topic_name` by given `key`
     /// and `partition` (if passed) with payload contains the `message`
-    /// data. Asynchronously waits until the message is delivered or the delivery
-    /// error occurred.
+    /// data. Asynchronously waits until the message is delivered or the delivery error occurred.
     ///
-    /// No payload data is copied. Method holds the data until message is
-    /// delivered.
+    /// No payload data is copied. Method holds the data until message is delivered.
     ///
-    /// Thread-safe and can be called from any number of threads
-    /// concurrently.
+    /// Thread-safe and can be called from any number of threads concurrently.
     ///
-    /// If `partition` not passed, partition is chosen by internal
-    /// Kafka partitioner.
+    /// 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.
+    /// @warning if `enable_idempotence` option is enabled, do not use both explicit partitions and Kafka-chosen ones.
     ///
     /// @throws SendException and its descendants if message is not delivered
     /// and acked by Kafka Broker in configured timeout.
     ///
-    /// @note Use SendException::IsRetryable method to understand whether there is
-    /// a sense to retry the message sending.
+    /// @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
     void Send(
         utils::zstring_view topic_name,
@@ -110,11 +107,40 @@ 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.
+    template <typename Messages>
+    std::enable_if_t<
+        std::is_convertible_v<decltype(std::declval<const Messages&>()[0]), std::string_view> &&
+        std::is_integral_v<decltype(std::declval<const Messages&>().size())> >
+    Send(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        const Messages& messages,
+        std::optional<std::uint32_t> partition = kUnassignedPartition,
+        HeaderViews headers = {}
+    ) const {
+        SendWrapper(topic_name, key, impl::MessagesAdapter{messages}, partition, std::move(headers));
+    }
+
     /// @brief Same as Producer::Send, but returns the task which can be
     /// used to wait the message delivery manually.
     ///
-    /// @warning If user schedules a batch of send requests with
-    /// Producer::SendAsync, some send
+    /// @warning If user schedules a batch of send requests with Producer::SendAsync, some send
     /// requests may be retried by the library (for instance, in case of network
     /// blink). Though, the order messages are written to partition may differ
     /// from the order messages are initially sent
@@ -141,6 +167,22 @@ class Producer final {
         impl::HeadersHolder&& headers_holder
     ) const;
 
+    void SendImpl(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        const impl::Messages& messages,
+        std::optional<std::uint32_t> partition,
+        std::vector<impl::HeadersHolder>&& headers_holders
+    ) const;
+
+    void SendWrapper(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        const impl::Messages& messages,
+        std::optional<std::uint32_t> partition,
+        HeaderViews headers
+    ) 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

@@ -138,20 +138,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,
+    const Messages& 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,34 +217,42 @@ 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"
 #endif
-    // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks,cppcoreguidelines-pro-type-const-cast)
-    const rd_kafka_resp_err_t enqueue_error = rd_kafka_producev(
-        producer_.GetHandle(),
-        RD_KAFKA_V_TOPIC(topic_name.c_str()),
-        RD_KAFKA_V_KEY(key.data(), key.size()),
-        RD_KAFKA_V_VALUE(const_cast<char*>(message.data()), message.size()),
-        RD_KAFKA_V_MSGFLAGS(0),
-        RD_KAFKA_V_HEADERS(headers_holder.GetHandle()),
-        RD_KAFKA_V_PARTITION(partition.value_or(RD_KAFKA_PARTITION_UA)),
-        RD_KAFKA_V_OPAQUE(waiter.get()),
-        RD_KAFKA_V_END
-    );
-    // NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks,cppcoreguidelines-pro-type-const-cast)
+        // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks,cppcoreguidelines-pro-type-const-cast)
+        const rd_kafka_resp_err_t enqueue_error = rd_kafka_producev(
+            producer_.GetHandle(),
+            RD_KAFKA_V_TOPIC(topic_name.c_str()),
+            RD_KAFKA_V_KEY(key.data(), key.size()),
+            RD_KAFKA_V_VALUE(const_cast<char*>(message.data()), message.size()),
+            RD_KAFKA_V_MSGFLAGS(0),
+            RD_KAFKA_V_HEADERS(headers_holder.GetHandle()),
+            RD_KAFKA_V_PARTITION(partition.value_or(RD_KAFKA_PARTITION_UA)),
+            RD_KAFKA_V_OPAQUE(waiter.get()),
+            RD_KAFKA_V_END
+        );
+        // NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks,cppcoreguidelines-pro-type-const-cast)
 
 #ifdef __clang__
 #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::Yield();
+            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,6 +6,8 @@
 
 #include <librdkafka/rdkafka.h>
 
+#include <userver/engine/deadline.hpp>
+#include <userver/kafka/impl/messages.hpp>
 #include <userver/kafka/impl/stats.hpp>
 #include <userver/logging/level.hpp>
 #include <userver/utils/periodic_task.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,
+        const Messages& 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.