feat docs: introduce Kafka tutorial

commit_hash:04d358d3fe66734f7633845b9717ee0b0c338ea7

.mapping.json

@@ -3310,6 +3310,7 @@
   "scripts/docs/en/userver/tutorial/hello_service.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/hello_service.md",
   "scripts/docs/en/userver/tutorial/http_caching.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/http_caching.md",
   "scripts/docs/en/userver/tutorial/json_to_yaml.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/json_to_yaml.md",
+  "scripts/docs/en/userver/tutorial/kafka_service.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/kafka_service.md",
   "scripts/docs/en/userver/tutorial/mongo_service.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/mongo_service.md",
   "scripts/docs/en/userver/tutorial/multipart_service.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/multipart_service.md",
   "scripts/docs/en/userver/tutorial/postgres_service.md":"taxi/uservices/userver/scripts/docs/en/userver/tutorial/postgres_service.md",

Makefile

@@ -16,7 +16,7 @@ CMAKE_RELEASE_FLAGS += -DCMAKE_BUILD_TYPE=Release $(CMAKE_COMMON_FLAGS)
 .PHONY: all
 all: test-debug test-release
 
-# Requires doxygen 1.9.8+
+# Requires doxygen 1.10.0+
 .PHONY: docs
 docs:
 	@rm -rf docs/*

kafka/functional_tests/integrational_tests/kafka_service.cpp

@@ -131,7 +131,7 @@ class HandlerKafkaProducers final
       const server::http::HttpRequest& request) const;
 
  private:
-  std::unordered_map<std::string, kafka::Producer&> producer_by_topic_;
+  std::unordered_map<std::string, const kafka::Producer&> producer_by_topic_;
 };
 
 }  // namespace functional_tests

kafka/include/userver/kafka/consumer_component.hpp

@@ -58,12 +58,16 @@ class Consumer;
 
 class ConsumerComponent final : public components::ComponentBase {
  public:
-  static constexpr std::string_view kName = "kafka-consumer";
+  /// @ingroup userver_component_names
+  /// @brief The default name of kafka::ConsumerComponent component
+  static constexpr std::string_view kName{"kafka-consumer"};
 
   ConsumerComponent(const components::ComponentConfig& config,
                     const components::ComponentContext& context);
   ~ConsumerComponent() override;
 
+  /// @brief Returns consumer instance.
+  /// @see kafka::ConsumerScope
   ConsumerScope GetConsumer();
 
   static yaml_config::Schema GetStaticConfigSchema();

kafka/include/userver/kafka/consumer_scope.hpp

@@ -27,11 +27,11 @@ class Consumer;
 ///
 /// Common usage:
 ///
-/// @snippet samples/kafka_service/main.cpp  Kafka service sample - consumer usage
+/// @snippet samples/kafka_service/src/consumer_handler.cpp  Kafka service sample - consumer usage
 ///
 /// ## Important implementation details
 ///
-/// ConsumerScope holds reference to `impl::Consumer` that actually
+/// ConsumerScope holds reference to kafka::impl::Consumer that actually
 /// represents the Apache Kafka Balanced Consumer.
 ///
 /// It exposes the API for asynchronous message batches processing that is

kafka/include/userver/kafka/exceptions.hpp

@@ -12,6 +12,10 @@ class SendException : public std::runtime_error {
  public:
   using std::runtime_error::runtime_error;
 
+  /// @brief Returns whether it makes sense to retry failed send.
+  ///
+  /// @see
+  /// https://docs.confluent.io/platform/current/clients/librdkafka/html/md_INTRODUCTION.html#autotoc_md8
   bool IsRetryable() const noexcept;
 
  protected:

kafka/include/userver/kafka/producer.hpp

@@ -88,7 +88,7 @@ class Producer final {
   ///
   /// @note Use SendException::IsRetryable method to understand whether there is
   /// a sense to retry the message sending.
-  /// @snippet kafka/tests/producer_test.cpp Producer retryable error
+  /// @snippet kafka/tests/producer_kafkatest.cpp Producer retryable error
   void Send(const std::string& topic_name, std::string_view key,
             std::string_view message,
             std::optional<std::uint32_t> partition = std::nullopt) const;
@@ -101,14 +101,14 @@ class Producer final {
   /// 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_test.cpp Producer batch send async
+  /// @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 = std::nullopt) const;
 
   /// @brief Dumps per topic messages produce statistics. No expected to be
   /// called manually.
-  /// @see impl/stats.hpp
+  /// @see kafka/impl/stats.hpp
   void DumpMetric(utils::statistics::Writer& writer) const;
 
  private:

kafka/include/userver/kafka/producer_component.hpp

@@ -50,13 +50,17 @@ namespace kafka {
 
 class ProducerComponent final : public components::ComponentBase {
  public:
-  static constexpr std::string_view kName = "kafka-producer";
+  /// @ingroup userver_component_names
+  /// @brief The default name of kafka::ProducerComponent component
+  static constexpr std::string_view kName{"kafka-producer"};
 
   ProducerComponent(const components::ComponentConfig& config,
                     const components::ComponentContext& context);
   ~ProducerComponent() override;
 
-  Producer& GetProducer();
+  /// @brief Returns a producer instance reference.
+  /// @see kafka::Producer
+  const Producer& GetProducer();
 
   static yaml_config::Schema GetStaticConfigSchema();
 

kafka/src/kafka/producer_component.cpp

@@ -35,7 +35,7 @@ ProducerComponent::ProducerComponent(
 
 ProducerComponent::~ProducerComponent() { statistics_holder_.Unregister(); }
 
-Producer& ProducerComponent::GetProducer() { return producer_; }
+const Producer& ProducerComponent::GetProducer() { return producer_; }
 
 yaml_config::Schema ProducerComponent::GetStaticConfigSchema() {
   return yaml_config::MergeSchemas<components::ComponentBase>(R"(

kafka/utest/include/userver/kafka/utest/kafka_fixture.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <chrono>
 #include <deque>
 #include <vector>
@@ -15,6 +16,8 @@ USERVER_NAMESPACE_BEGIN
 
 namespace kafka::utest {
 
+/// @ingroup userver_utest
+///
 /// @brief Message owning data wrapper for unit tests.
 struct Message {
   std::string topic;
@@ -25,6 +28,8 @@ struct Message {
 
 bool operator==(const Message& lhs, const Message& rhs);
 
+/// @ingroup userver_utest
+///
 /// @brief Helper for Kafka unit testing.
 ///
 /// KafkaCluster is useful for
@@ -80,16 +85,20 @@ class KafkaCluster : public ::testing::Test {
       std::size_t count, std::function<std::string(std::size_t)> nameGenerator,
       impl::ProducerConfiguration configuration = {});
 
+  /// @brief Creates temporary producer and send messages span.
   void SendMessages(utils::span<const Message> messages);
 
   impl::Consumer MakeConsumer(const std::string& name,
                               const std::vector<std::string>& topics,
                               impl::ConsumerConfiguration configuration = {},
                               impl::ConsumerExecutionParams params = {});
 
-  std::vector<Message> ReceiveMessages(impl::Consumer& consumer,
-                                       std::size_t expected_messages_count,
-                                       bool commit_after_receive = true);
+  /// @brief Starts consumer, waits until `expected_message_count` messages are
+  /// consumed, calls `user_callback` if set, stops consumer.
+  std::vector<Message> ReceiveMessages(
+      impl::Consumer& consumer, std::size_t expected_messages_count,
+      bool commit_after_receive = true,
+      std::optional<std::function<void(MessageBatchView)>> user_callback = {});
 
  private:
   impl::Secret AddBootstrapServers(impl::Secret secrets) const;

kafka/utest/src/kafka/utest/kafka_fixture.cpp

@@ -147,19 +147,24 @@ impl::Consumer KafkaCluster::MakeConsumer(
 
 std::vector<Message> KafkaCluster::ReceiveMessages(
     impl::Consumer& consumer, std::size_t expected_messages_count,
-    bool commit_after_receive) {
+    bool commit_after_receive,
+    std::optional<std::function<void(MessageBatchView)>> user_callback) {
   std::vector<Message> received_messages;
 
   engine::SingleUseEvent event;
   auto consumer_scope = consumer.MakeConsumerScope();
   consumer_scope.Start(
       [&received_messages, expected_messages_count, &event, &consumer_scope,
+       &user_callback,
        commit = commit_after_receive](MessageBatchView messages) {
         for (const auto& message : messages) {
           received_messages.push_back(Message{
               message.GetTopic(), std::string{message.GetKey()},
               std::string{message.GetPayload()}, message.GetPartition()});
         }
+        if (user_callback) {
+          (*user_callback)(messages);
+        }
         if (commit) {
           consumer_scope.AsyncCommit();
         }

samples/kafka_service/CMakeLists.txt

@@ -19,6 +19,7 @@ target_include_directories(${PROJECT_NAME}_objs PUBLIC src/)
 add_executable(${PROJECT_NAME} "main.cpp")
 target_link_libraries(${PROJECT_NAME} PRIVATE ${PROJECT_NAME}_objs)
 
+# /// [Kafka service sample - kafka unit test cmake]
 add_executable(${PROJECT_NAME}-unittest "unittest/kafka_test.cpp")
 target_link_libraries(${PROJECT_NAME}-unittest
     PRIVATE ${PROJECT_NAME}_objs userver::kafka-utest)
@@ -31,9 +32,14 @@ userver_add_utest(
     "TESTSUITE_KAFKA_CUSTOM_TOPICS=test-topic-1:1,test-topic-2:1"
 )
 
+# /// [Kafka service sample - kafka unit test cmake]
+
+# /// [Kafka service sample - kafka functional test cmake]
 userver_testsuite_add_simple(
     TEST_ENV
     "TESTSUITE_KAFKA_SERVER_START_TIMEOUT=120.0"
     "TESTSUITE_KAFKA_SERVER_HOST=[::1]"
     "TESTSUITE_KAFKA_CUSTOM_TOPICS=test-topic-1:1,test-topic-2:1"
 )
+
+# /// [Kafka service sample - kafka functional test cmake]

samples/kafka_service/main.cpp

@@ -17,6 +17,7 @@
 #include <consumer_handler.hpp>
 #include <producer_handler.hpp>
 
+/// [Kafka service sample - main]
 int main(int argc, char* argv[]) {
   const auto components_list =
       components::MinimalServerComponentList()
@@ -33,3 +34,4 @@ int main(int argc, char* argv[]) {
 
   return utils::DaemonMain(argc, argv, components_list);
 }
+/// [Kafka service sample - main]

samples/kafka_service/src/consume.cpp

@@ -5,6 +5,7 @@
 
 namespace kafka_sample {
 
+/// [Kafka service sample - consume]
 void Consume(kafka::MessageBatchView messages) {
   for (const auto& message : messages) {
     if (!message.GetTimestamp().has_value()) {
@@ -19,5 +20,6 @@ void Consume(kafka::MessageBatchView messages) {
     }());
   }
 }
+/// [Kafka service sample - consume]
 
 }  // namespace kafka_sample

samples/kafka_service/src/consumer_handler.hpp

@@ -2,14 +2,19 @@
 
 #include <string_view>
 
+/// [Kafka service sample - consumer include]
+
 #include <userver/kafka/consumer_scope.hpp>
 
+/// [Kafka service sample - consumer include]
+
 #include <userver/components/component_base.hpp>
 
 #include <userver/utest/using_namespace_userver.hpp>
 
 namespace kafka_sample {
 
+/// [Kafka service sample - consumer component declaration]
 class ConsumerHandler final : public components::ComponentBase {
  public:
   static constexpr std::string_view kName{"consumer-handler"};
@@ -21,5 +26,6 @@ class ConsumerHandler final : public components::ComponentBase {
   // Subscriptions must be the last fields! Add new fields above this comment.
   kafka::ConsumerScope consumer_;
 };
+/// [Kafka service sample - consumer component declaration]
 
 }  // namespace kafka_sample

samples/kafka_service/src/produce.cpp

@@ -4,6 +4,7 @@
 
 namespace kafka_sample {
 
+/// [Kafka service sample - produce]
 SendStatus Produce(const kafka::Producer& producer,
                    const RequestMessage& message) {
   try {
@@ -14,5 +15,6 @@ SendStatus Produce(const kafka::Producer& producer,
                             : SendStatus::kErrorNonRetryable;
   }
 }
+/// [Kafka service sample - produce]
 
 }  // namespace kafka_sample

samples/kafka_service/src/produce.hpp

@@ -14,6 +14,7 @@ enum class SendStatus {
   kErrorNonRetryable,
 };
 
+/// @brief Example message data.
 struct RequestMessage {
   std::string topic;
   std::string key;

samples/kafka_service/src/producer_handler.cpp

@@ -42,12 +42,15 @@ bool IsCorrectRequest(const formats::json::Value& request_json) {
 
 }  // namespace
 
+/// [Kafka service sample - producer component find]
 ProducerHandler::ProducerHandler(const components::ComponentConfig& config,
                                  const components::ComponentContext& context)
     : server::handlers::HttpHandlerJsonBase{config, context},
       producer_{
           context.FindComponent<kafka::ProducerComponent>().GetProducer()} {}
+/// [Kafka service sample - producer component find]
 
+/// [Kafka service sample - producer handler implementation]
 formats::json::Value ProducerHandler::HandleRequestJsonThrow(
     const server::http::HttpRequest& request,
     const formats::json::Value& request_json,
@@ -71,5 +74,6 @@ formats::json::Value ProducerHandler::HandleRequestJsonThrow(
   }
   UINVARIANT(false, "Unknown produce status");
 }
+/// [Kafka service sample - producer handler implementation]
 
 }  // namespace kafka_sample

samples/kafka_service/src/producer_handler.hpp

@@ -2,14 +2,19 @@
 
 #include <string_view>
 
+/// [Kafka service sample - producer include]
+
 #include <userver/kafka/producer.hpp>
 
+/// [Kafka service sample - producer include]
+
 #include <userver/server/handlers/http_handler_json_base.hpp>
 
 #include <userver/utest/using_namespace_userver.hpp>
 
 namespace kafka_sample {
 
+/// [Kafka service sample - producer component declaration]
 class ProducerHandler final : public server::handlers::HttpHandlerJsonBase {
  public:
   static constexpr std::string_view kName{"producer-handler"};
@@ -23,7 +28,8 @@ class ProducerHandler final : public server::handlers::HttpHandlerJsonBase {
       server::request::RequestContext& context) const override;
 
  private:
-  kafka::Producer& producer_;
+  const kafka::Producer& producer_;
 };
+/// [Kafka service sample - producer component declaration]
 
 }  // namespace kafka_sample

samples/kafka_service/testsuite/conftest.py

@@ -1,12 +1,27 @@
 import pytest
 
+# /// [Kafka service sample - kafka testsuite include]
+
 pytest_plugins = ['pytest_userver.plugins.kafka']
 
+# /// [Kafka service sample - kafka testsuite include]
+
 
 # /// [Kafka service sample - secdist]
 @pytest.fixture(scope='session')
 def service_env(kafka_secdist) -> dict:
-    """kafka_secist fixture generates the secdist config"""
+    """
+    Note: kafka_secist fixture generates the secdist config
+
+    Expected secdist format is:
+    "kafka_settings": {
+        "<kafka-component-name>": {
+            "brokers": "<brokers comma-separated endpoint list>",
+            "username": "SASL2 username (may be empty if use PLAINTEXT)",
+            "password": "SASL2 password (may be empty if use PLAINTEXT)"
+        }
+    }
+    """
 
     return {'SECDIST_CONFIG': kafka_secdist}
     # /// [Kafka service sample - secdist]