feat kafka: add bulk send producer method by disaykin · Pull Request #1134 · userver-framework/userver

disaykin · 2026-03-06T11:18:02Z

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

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

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

Note: by creating a PR or an issue you automatically agree to the CLA. See CONTRIBUTING.md. Feel free to remove this note, the agreement holds.

disaykin · 2026-03-16T05:40:40Z

@apolukhin Что нужно сделать, чтобы запустить тестовый пайплайн?

disaykin · 2026-03-16T19:46:21Z

Can somebody help me?

apolukhin

Nice! A few comments follow

apolukhin · 2026-03-17T09:57:13Z

kafka/include/userver/kafka/exceptions.hpp

+public:
+    using ExceptionMap = std::map<std::size_t, std::exception_ptr>;
+
+    BulkSendException(ExceptionMap nested_exceptions);


apolukhin · 2026-03-17T09:57:57Z

kafka/include/userver/kafka/producer.hpp

+    void Send(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        utils::span<const std::string> messages,


utils::span<const std::string_view> messages ?

Сначала так и сделал, но потом не смог победить. Нужно создать массив вьюх на временные строки, чтобы потом он эффективно преобразовался в span. Временные строки разрушатся до того, как мы сможем передать их в метод Send. Если подскажете, как сделать - с удовольствием переделаю.

попробовал сделать, чтобы метод Send принимал на вход любой контейнер, имеющий метод size() и оператор индексирования, и который хранит элементы, преобразуемые к std::string_view, так, чтобы это было валидно на C++-17 и чтобы не вытаскивать кишки реализации наружу. Получилось не очень изысканно, прошу совета от гуру, как это лучше сделать

apolukhin · 2026-03-17T09:58:20Z

kafka/include/userver/kafka/producer.hpp

+    void SendImpl(
+        utils::zstring_view topic_name,
+        std::string_view key,
+        utils::span<const std::string> messages,


utils::span<const std::string_view> messages ?

аналогично, не знаю, как эффективно преобразовать std::vector<std::string> в utils::span<std::string_view>

apolukhin · 2026-03-17T10:01:36Z

kafka/src/kafka/impl/producer_impl.cpp

+	} 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});


This is too much... How about just an engine::Yield()?

apolukhin · 2026-03-17T10:04:37Z

kafka/tests/producer_kafkatest.cpp


-    UEXPECT_THROW(make_send_request(), kafka::QueueFullException);
-
-    /// [Producer retryable error]


This example is used in docs as a sample userver/kafka/include/userver/kafka/producer.hpp. Please do not remove it

Беда, не знал. Как в этом случае поступить? Я переделал сценарий так, чтобы в случае переполнения кода Producer::Send повторял попытки положить сообщение в очередь вплоть до исчерпания delivery_timeout. Либо я добавляю опцию, чтобы Send одного сообщения сразу возвращал ошибку, что очередь полна, либо нужно править документацию.

Вернул старое поведение для одиночного Send, чтобы тест и комментарий остались валидными. Других восстановимых ошибок (кроме DeliveryTimeoutError) юсервером не поддерживается, поэтому либо внедряем пример из теста в документацию для демонстрации, как обрабатывать IsRetriable, либо мокаем ProducerImpl, чтобы он первое время возвращал DeliveryTimeoutError, либо возвращаем старое поведение. Пока предпочел третий вариант.

Всё-таки решил переделать на новый лад. Пример использования IsRetriable() вынес из теста в документацию

apolukhin · 2026-03-17T10:06:14Z

@apolukhin Что нужно сделать, чтобы запустить тестовый пайплайн?

Сначала отревьюил, потом только заметил что можно писать на русском :)

Запустил тестовый пайплайн

disaykin · 2026-03-17T19:28:24Z

при установке кафки на ubuntu 24.04 и ubuntu 24.04:

tar: This does not look like a tar archive
gzip: stdin: not in gzip format
tar: Child returned status 1
tar: Error is not recoverable: exiting now
cp: cannot stat '/etc/kafka/kafka_2.13-4.0.0/*': No such file or directory

интересно, почему это не привело к фэйлу пайплайна?

Это приводит к ошибкам на этапе функциональных тестов кафки:

2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: Current Java version is 17
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: 
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: Kafka sources not found in KAFKA_HOME and in /etc/kafka !!!.
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: Please download Kafka from https://kafka.apache.org/downloads,
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: unpack the archive and place path to it in KAFKA_HOME.
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: Note: For MacOS just install KAFKA with 'brew install kafka'
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: 
2: ERROR    testsuite.environment.shell:shell.py:70 env/kafka/stop[49366]: Kafka is not found
2: ERROR    testsuite.environment.shell:shell.py:71 env/kafka/stop[49366]: subprocess ['/home/runner/work/userver/userver/build_debug/venv-userver-testenv/lib/python3.10/site-packages/testsuite/databases/kafka/scripts/service-kafka', 'stop'] exited with code 1

disaykin · 2026-03-17T20:28:57Z

На ubuntu-24.04: gcc-13 + release сбойнул не связанный с моими изменениями тест:

[ RUN      ] HttpClientProxy.HttpsTruncatedBody
userver/core/src/clients/http/client_crl_test.cpp:888: Failure
  Expected: 'response_future.Get()' throws 'userver::v2_16_rc::clients::http::NetworkProblemException'.
  Actual: it throws an exception of a different type 'userver::v2_16_rc::clients::http::TechnicalError' with message 'Technical error, curl error: Transferred a partial file, url: https://[::1]:33411/'. 

[  FAILED  ] HttpClientProxy.HttpsTruncatedBody (133 ms)

disaykin · 2026-03-18T13:55:42Z

@apolukhin нужна помощь с воспроизведением локальной сборки с тестами. Локально на ноуте с Ubuntu 24.04 интеграционные тесты на Кафку не проходят одинаково, что с моими изменениями, что без них

disaykin · 2026-03-18T13:57:56Z

отломалась компиляция на ubuntu-minimal:

/home/runner/work/userver/userver/core/src/server/middlewares/graceful_shutdown_headers.cpp: In member function ‘virtual void userver::v2_16_rc::server::middlewares::GracefulShutdownHeaders::HandleRequest(userver::v2_16_rc::server::http::HttpRequest&, userver::v2_16_rc::server::request::RequestContext&) const’:
/home/runner/work/userver/userver/core/src/server/middlewares/graceful_shutdown_headers.cpp:27:62: error: ‘join’ is not a member of ‘fmt’
   27 |                 response.SetHeader(name, fmt::to_string(fmt::join(values, ", ")));
      |                                                              ^~~~
At global scope:
cc1plus: note: unrecognized command-line option ‘-Wno-gnu-zero-variadic-macro-arguments’ may have been intended to silence earlier diagnostics
gmake[2]: *** [core/CMakeFiles/userver-core.dir/build.make:4830: core/CMakeFiles/userver-core.dir/src/server/middlewares/graceful_shutdown_headers.cpp.o] Error 1
gmake[2]: *** Waiting for unfinished jobs....
gmake[1]: *** [CMakeFiles/Makefile2:2703: core/CMakeFiles/userver-core.dir/all] Error 2
gmake: *** [Makefile:146: all] Error 2
[136](https://github.com/userver-framework/userver/actions/runs/23245973355/job/67574368337?pr=1134#step:9:1363)

apolukhin · 2026-03-18T14:33:25Z

отломалась компиляция на ubuntu-minimal:

/home/runner/work/userver/userver/core/src/server/middlewares/graceful_shutdown_headers.cpp: In member function ‘virtual void userver::v2_16_rc::server::middlewares::GracefulShutdownHeaders::HandleRequest(userver::v2_16_rc::server::http::HttpRequest&, userver::v2_16_rc::server::request::RequestContext&) const’:
/home/runner/work/userver/userver/core/src/server/middlewares/graceful_shutdown_headers.cpp:27:62: error: ‘join’ is not a member of ‘fmt’
   27 |                 response.SetHeader(name, fmt::to_string(fmt::join(values, ", ")));
      |                                                              ^~~~
At global scope:
cc1plus: note: unrecognized command-line option ‘-Wno-gnu-zero-variadic-macro-arguments’ may have been intended to silence earlier diagnostics
gmake[2]: *** [core/CMakeFiles/userver-core.dir/build.make:4830: core/CMakeFiles/userver-core.dir/src/server/middlewares/graceful_shutdown_headers.cpp.o] Error 1
gmake[2]: *** Waiting for unfinished jobs....
gmake[1]: *** [CMakeFiles/Makefile2:2703: core/CMakeFiles/userver-core.dir/all] Error 2
gmake: *** [Makefile:146: all] Error 2
[136](https://github.com/userver-framework/userver/actions/runs/23245973355/job/67574368337?pr=1134#step:9:1363)

Фикс уже в пути... Там не хватает inlude fmt/ranges.h или чего-то подобного

Можете просто взять версию userver до комита 332dcb5

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

disaykin force-pushed the kafka-send-batch branch 5 times, most recently from 31b4214 to 79d1485 Compare March 10, 2026 19:34

apolukhin requested changes Mar 17, 2026

View reviewed changes

disaykin force-pushed the kafka-send-batch branch from 79d1485 to bc9c629 Compare March 18, 2026 13:02

disaykin force-pushed the kafka-send-batch branch 2 times, most recently from b072ef9 to f6113f4 Compare March 20, 2026 06:04

disaykin added 3 commits March 20, 2026 09:04

Unify and simplify Kafka installation scripts

0202a4a

WiP

20db89b

disaykin force-pushed the kafka-send-batch branch from f6113f4 to 20db89b Compare March 20, 2026 06:04

WiP

6658379


		UEXPECT_THROW(make_send_request(), kafka::QueueFullException);

		/// [Producer retryable error]

Conversation

disaykin commented Mar 6, 2026

Uh oh!

disaykin commented Mar 16, 2026

Uh oh!

disaykin commented Mar 16, 2026

Uh oh!

apolukhin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

disaykin Mar 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

disaykin Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

disaykin Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apolukhin commented Mar 17, 2026

Uh oh!

disaykin commented Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

disaykin commented Mar 17, 2026

Uh oh!

disaykin commented Mar 18, 2026

Uh oh!

disaykin commented Mar 18, 2026

Uh oh!

apolukhin commented Mar 18, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

disaykin Mar 18, 2026 •

edited

Loading

disaykin Mar 17, 2026 •

edited

Loading

disaykin Mar 17, 2026 •

edited

Loading

disaykin commented Mar 17, 2026 •

edited

Loading