producer.hpp

#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>
 
USERVER_NAMESPACE_BEGIN
 
namespace kafka {
 
/// @brief Unassigned partition.
///
/// The unassigned partition is used by the producer API for messages
/// that should be partitioned using the default partitioner.
///
/// @note By default partitions are distributed uniformly.
extern const std::optional<std::uint32_t> kUnassignedPartition;
 
namespace impl {
 
class Configuration;
class ProducerImpl;
 
struct ProducerConfiguration;
struct Secret;
 
class HeadersHolder;
 
}  // namespace impl
 
/// @ingroup userver_clients
///
/// @brief Apache Kafka Producer Client.
///
/// It exposes the API to send an arbitrary message to the Kafka Broker to
/// certain topic by given key and optional partition.
///
/// ## Important implementation details
///
/// Message send tasks handling other messages' delivery reports, suspending
/// their execution while no events exist.
/// This makes message production parallel and leads to high Producer
/// scalability.
///
/// Producer maintains per topic statistics including the broker
/// connection errors.
///
/// @remark Destructor may wait for no more than a 2 x `delivery_timeout` to
/// ensure all sent messages are properly delivered.
///
/// @see https://docs.confluent.io/platform/current/clients/producer.html
class Producer final {
public:
    /// @cond
    // @param producer_task_processor is task processor where producer creates
    // tasks for message delivery scheduling and waiting.
    Producer(
        const std::string& name,
        engine::TaskProcessor& producer_task_processor,
        const impl::ProducerConfiguration& configuration,
        const impl::Secret& secrets
    );
    /// @endcond
 
    /// @brief Waits until all messages are sent for at most 2 x
    /// `delivery_timeout` and destroys the producer.
    ///
    /// @remark In a basic producer use cases, the destructor returns immediately.
    ~Producer();
 
    Producer(const Producer&) = delete;
    Producer(Producer&&) = delete;
 
    Producer& operator=(const Producer&) = delete;
    Producer& operator=(Producer&&) = delete;
 
    /// @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.
    ///
    /// 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.
    ///
    /// 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 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.
    /// @snippet kafka/tests/producer_kafkatest.cpp Producer retryable error
    void Send(
        utils::zstring_view topic_name,
        std::string_view key,
        std::string_view message,
        std::optional<std::uint32_t> partition = kUnassignedPartition,
        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
    /// 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
    /// @snippet kafka/tests/producer_kafkatest.cpp Producer batch send async
    [[nodiscard]] engine::TaskWithResult<void> SendAsync(
        std::string topic_name,
        std::string key,
        std::string message,
        std::optional<std::uint32_t> partition = kUnassignedPartition,
        HeaderViews headers = {}
    ) const;
 
    /// @brief Dumps per topic messages produce statistics. No expected to be
    /// called manually.
    /// @see kafka/impl/stats.hpp
    void DumpMetric(utils::statistics::Writer& writer) const;
 
private:
    void SendImpl(
        utils::zstring_view topic_name,
        std::string_view key,
        std::string_view message,
        std::optional<std::uint32_t> partition,
        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_;
 
    static constexpr std::size_t kImplSize{944};
    static constexpr std::size_t kImplAlign{16};
    utils::FastPimpl<impl::ProducerImpl, kImplSize, kImplAlign> producer_;
};
 
}  // namespace kafka
 
USERVER_NAMESPACE_END