--- title: "Модуль Kafka: продюсер и консьюмер" description: "Модуль testcontainers kafka: брокер KRaft без Zookeeper — kafka.Run, адреса через Brokers(ctx), создание топика, продюсер и консьюмер в тесте." course: "Testcontainers в Go: интеграционные тесты с Docker" url: "https://golangmaster.ru/courses/testcontainers/lessons/08-kafka-module/" published: "2026-09-19" updated: "2026-09-21" minutes: 18 --- # Модуль Kafka: продюсер и консьюмер *Курс «Testcontainers в Go: интеграционные тесты с Docker», урок 8 из 25. ~18 минут. Обновлено 21.09.2026. Источник: https://golangmaster.ru/courses/testcontainers/lessons/08-kafka-module/* kafka тесты go чаще всего обманывают сами себя: мок-интерфейс обещает доставку, но не знает ни про партиции, ни про оффсеты, ни про порядок сообщений. Модуль testcontainers kafka поднимает в Docker настоящий брокер в KRaft-режиме — без Zookeeper — и возвращает управление, только когда брокер действительно готов принимать клиентов. Вместо GenericContainer с ручным набором переменных окружения и wait-стратегий вы пишете один вызов kafka.Run и получаете типизированный KafkaContainer: адреса брокеров через Brokers(ctx), идентификатор кластера в поле ClusterID. Дальше — обычная работа клиентской библиотеки: создать топик, отправить сообщение продюсером, прочитать его консьюмером и убедиться, что оно дошло без потерь. В уроке соберём такой тест с нуля: каркас запуска, явное создание топика, тест «сообщение дошло» и грабли первого дня — долгий старт образа, порт 9093 вместо 9092 и партиции, из-за которых сообщения приходят не по порядку. Все вызовы урока сверены с исходниками модуля: увидим и то, что происходит внутри контейнера, пока Run ещё не вернулся. Брокер сообщений — второй по частоте гость интеграционных тестов после базы: заказы, уведомления, события между сервисами. Заменить его моком просто, но тогда тест не проверяет почти ничего из того, ради чего Kafka берут в проект: сериализацию значения, раскладку по партициям, поведение консьюмер-групп, порядок доставки. Голый GenericContainer тоже подходит плохо. У Kafka несколько листенеров — для клиентов, для обмена между брокерами и для контроллера, — advertised listeners обязаны указывать на адрес, доступный с хоста, а готовность брокера видно только по конкретной строке в логах. Собирать эту конструкцию руками в каждом пакете — верный способ получить флейк. Модуль kafka закрывает всё разом: выставляет переменные KRaft-режима, форматирует хранилище на старте, переписывает advertised listeners на маппированный адрес и возвращает типизированный контейнер с готовыми методами. Внутри — тот же GenericContainer и wait-стратегии из первых уроков курса, только собранные один раз в исходниках модуля, а не в каждом вашем тесте. Это та же экономия, что и с базой: специфика брокера описана один раз, а тесты остаются честными — с настоящим протоколом, настоящими оффсетами и настоящей доставкой. ### Почему testcontainers kafka, а не GenericContainer GenericContainer из урока про любой образ умеет поднять что угодно, но про Kafka не знает ничего: переменные KAFKA_*, три листенера, стартовый скрипт, ожидание готовности — всё руками, и ошибка в одном звене даёт тест, который висит до таймаута. Модуль выставляет весь набор за вас: роли процесса broker и controller, номер единственного брокера, факторы репликации внутренних топиков, равные единице, нулевую задержку ребалансировки групп. На одиночном брокере это убирает целые классы ошибок: без единичных факторов репликации брокер отказался бы создавать внутренние топики, а пауза на ребалансировку замедляла бы каждый прогон. Отдельная забота модуля — адресация: стартовый скрипт переписывает advertised listeners на маппированный адрес, поэтому клиенты с хоста подключаются без ручных правок конфигурации. Взамен вы получаете типизированный KafkaContainer: метод Brokers для адресов и поле ClusterID с идентификатором кластера. Кластера, впрочем, не будет: модуль поднимает ровно один брокер, опции про число брокеров у Run нет, а Brokers честно возвращает срез с единственным элементом — так и написано в документации метода. ### Каркас: kafka.Run и Brokers(ctx) ```go import ( "context" "log" "testing" "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/modules/kafka" ) func TestMain(m *testing.M) { ctx := context.Background() kc, err := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0", kafka.WithClusterID("test-cluster"), ) if err != nil { log.Fatalf("kafka не поднялся: %v", err) } defer testcontainers.TerminateContainer(kc) brokers, err := kc.Brokers(ctx) if err != nil { log.Fatalf("brokers: %v", err) } log.Printf("broker=%s cluster=%s", brokers[0], kc.ClusterID) m.Run() } ``` TestMain поднимает брокер один раз на пакет: Run принимает образ первым аргументом, опции модуля следом. kafka.WithClusterID записывает переменную CLUSTER_ID в окружение контейнера, а после старта Run читает её обратно и кладёт в поле ClusterID. Brokers возвращает срез адресов с одним элементом — хост и маппированный порт. TerminateContainer в defer уберёт контейнер при любом исходе, m.Run запускает тесты, и каждый пользуется готовым адресом. ### kafka.Run: KRaft без Zookeeper KRaft — режим Kafka, в котором координацией кластера занимается сам брокер: Zookeeper не нужен ни как отдельный контейнер, ни как зависимость в списке сервисов. Модуль готовит его сам: убирает из скрипта конфигурации переменную KAFKA_ZOOKEEPER_CONNECT и форматирует хранилище утилитой kafka-storage со случайным идентификатором до первого запуска брокера. Про версии модуль тоже сторожит: для официального образа confluent-local тег ниже 7.4.0 отвергается сразу ошибкой про KRaft — в таких версиях режима ещё не было, и контейнер даже не стартует; чужие образы проходят без проверки тега, но и гарантий конфигурации под них нет. Из дефолтов запомните три: факторы репликации внутренних топиков равны единице — иначе одиночный брокер не справится, нулевая задержка ребалансировки — консьюмер-группа подключается без паузы, и отключённый по числу сообщений flush — записи не сбрасываются на диск ради скорости прогона. kafka тесты go проходят на честном протоколе брокера, и все эти дефолты существуют ради скорости и предсказуемости теста, а не для галочки в документации. Если интересно, что именно настраивает модуль, откройте docker logs контейнера: роли процесса, форматирование хранилища и листенеры видны там в терминах самого брокера. > **Живое место: наружу 9093, а не 9092.** Наберите docker port у работающего контейнера — увидите 9093, хотя в статьях про Kafka привычнее 9092. Схема листенеров такая: PLAINTEXT для клиентов слушает 9093, BROKER для обмена между брокерами — 9092 и наружу не выходит, CONTROLLER для KRaft — 9094. Brokers(ctx) собирает адрес именно по 9093. Спорное мнение с условием: можно выставить 9092 наружу и подключаться к нему напрямую — но незачем: стартовый скрипт модуля уже переписал advertised listeners на маппированный адрес 9093, и клиенты с хоста ходят через него; ручной проброс 9092 в лучшем случае повторит готовое, а в худшем сломает адресацию, если однажды соберёте несколько брокеров сами. ### Создание топика: явно или авто-созданием Модуль топики не создаёт: его зона ответственности — поднять брокер и отдать адрес. Дальше два пути. Первый — авто-создание: брокер создаёт топик при первой отправке, и тест из исходников модуля так и живёт — шлёт сообщение в несуществующий топик и читает его. Быстро, но с сюрпризами: конфигурацию топика решает брокер по своим дефолтам, а момент создания попадает в середину теста. Второй путь — явный: админ-API клиентской библиотеки. В kafka-go это Client с адресом брокера и CreateTopicsRequest со списком TopicConfig: имя топика, число партиций, фактор репликации. На одиночном брокере репликация больше единицы не поднимется — живых брокеров не хватит. Явное создание дешевле отладки: тест ловит ошибку конфигурации на первом шаге, а не гадает на первой отправке, и получает топик с нулевыми оффсетами — ровно то состояние, что нужно для проверки доставки. Клиентская сторона kafka-go заслуживает отдельного разговора: [курс kafka-go](/courses/kafka-go/) разбирает Writer, Reader и админ-клиент по косточкам, здесь нам хватит трёх вызовов. ### Тест «сообщение дошло» ```go // brokers, ctx — из TestMain; нужны kafka-go и time client := &kafka.Client{Addr: kafka.TCP(brokers...)} _, err = client.CreateTopics(ctx, &kafka.CreateTopicsRequest{ Topics: []kafka.TopicConfig{ {Topic: "orders", NumPartitions: 1, ReplicationFactor: 1}, }, }) if err != nil { t.Fatalf("create topic: %v", err) } reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, Topic: "orders", }) defer reader.Close() writer := &kafka.Writer{Addr: kafka.TCP(brokers...), Topic: "orders"} defer writer.Close() if err = writer.WriteMessages(ctx, kafka.Message{ Key: []byte("order-1"), Value: []byte("paid"), }); err != nil { t.Fatalf("write: %v", err) } readCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() msg, err := reader.ReadMessage(readCtx) if err != nil { t.Fatalf("read: %v", err) } if string(msg.Key) != "order-1" { t.Fatalf("key = %s, want order-1", msg.Key) } ``` CreateTopics заводит топик orders с одной партицией и репликацией 1 — на одиночном брокере иначе нельзя. Читатель без GroupID привязан к одной партиции — нулевой по умолчанию — и стартует с FirstOffset: в исходниках kafka-go оффсет нового Reader инициализируется первым. Writer шлёт сообщение с ключом синхронно, ReadMessage блокируется до доставки или дедлайна: без него упавший продюсер подвесил бы тест до общего таймаута пакета. Сверка ключа — и сообщение дошло. ### Продюсер и консьюмер: WriteMessages и ReadMessage Writer — структура с адресом брокеров и именем топика: WriteMessages принимает любое число сообщений и возвращает ошибку, только если доставить не удалось. Ключ сообщения определяет партицию: с одной партицией порядок сквозной, с несколькими — лишь внутри каждой. Reader бывает двух режимов, и Validate не даст их перепутать: либо номер партиции, либо GroupID — оба сразу нельзя. Режим без группы — это чтение одной партиции без коммитов: состояние оффсета держите вы, и для теста с единственной партицией это самый короткий путь. Групповой режим берёт на себя ребалансировку и коммиты, но добавляет время координации: группа должна собраться, прежде чем читатель получит свои партиции. Дисциплина теста от режима не зависит: сначала читатель, потом отправка — тест в исходниках модуля дожидается готовности консьюмера до первого сообщения. ReadMessage блокируется, поэтому дедлайн обязателен: контекст с таймаутом превращает молчание брокера в красный тест через секунды, а не через десять минут. Возвращённый Message несёт топик, партицию, оффсет, ключ и значение — сравнивайте с ожидаемым то, что входит в контракт сервиса. Порядок читатель-потом-отправка стоит соблюдать и по другой причине: он делает тест независимым от стартовых оффсетов конкретной библиотеки. ### Проверка: адреса брокеров **Вопрос:** Тест вызвал kc.Brokers(ctx) у контейнера из kafka.Run. Что вернёт метод? - ✅ Срез с одним адресом: хост и маппированный порт 9093 - Срез с тремя адресами: по одному на каждый листенер - Строку localhost:9092 — стандартный порт брокера Kafka - Пустой срез: адреса узнаёт только клиентская библиотека Brokers собирает адрес через PortEndpoint по порту 9093/tcp — единственному, который модуль выставляет наружу; 9092 и 9094 остаются внутренними. Возвращается срез, а не строка: формат под общий случай со многими брокерами, но модуль поднимает один, и срез содержит единственный элемент — это зафиксировано в документации метода. ### Задача → приём: Kafka в тесте | Задача | Приём | Пример | | --- | --- | --- | | Поднять брокер KRaft | kafka.Run: образ обязателен, опции модуля следом | kafka.Run(ctx, "confluentinc/confluent-local:7.5.0") | | Задать идентификатор кластера | WithClusterID пишет CLUSTER_ID; поле ClusterID прочитает обратно | kafka.WithClusterID("test-cluster") | | Узнать адрес брокера | Brokers(ctx) — срез с одним адресом на порту 9093 | brokers, err := kc.Brokers(ctx) | | Создать топик | админ-API клиента или авто-создание брокером | client.CreateTopics(ctx, req) | | Отправить сообщение | Writer.WriteMessages: ключ выбирает партицию | writer.WriteMessages(ctx, msg) | | Прочитать сообщение | Reader.ReadMessage с дедлайном контекста | reader.ReadMessage(readCtx) | | Убрать контейнер | CleanupContainer в тесте или TerminateContainer в defer | testcontainers.CleanupContainer(t, kc) | ### Грабли: таймауты старта и партиции Первая грабля — старт. Образ confluent-local тяжёлый: первый прогон качает сотни мегабайт, затем модуль ждёт маппированный порт, копирует стартовый скрипт в контейнер и ждёт строку про переход из RECOVERY в RUNNING в логах; на холодной CI-машине это десятки секунд. Короткий context.WithTimeout вокруг Run убьёт контейнер посреди старта — дедлайн ставят на весь прогон или не ставят вовсе, а образ кешируется между запусками. Вторая грабля — партиции. Топик на одиночном брокере живёт с одной партицией и репликацией 1: больше брокер не прокормит. Несколько партиций ломают порядок: ключи раскладываются по партициям, и два сообщения одного сценария приходят читателю в порядке внутри партиции, а не внутри сценария. Читатель без группы привязан к нулевой партиции — при нескольких он увидит её одну и решит, что сообщения потерялись. Третья — много брокеров: модуль их не собирает, это несколько контейнеров в общей сети, [тема урока про сети](/courses/testcontainers/lessons/14-seti/). ### Чек-лист Kafka в тестах - kafka.Run вместо GenericContainer: KRaft-переменные, стартовый скрипт и ожидание готовности уже настроены - образ confluent-local первым аргументом; тег ниже минимума KRaft модуль отвергнет ошибкой до старта - kafka.WithClusterID — идентификатор кластера; он же вернётся в поле ClusterID - адрес брокера — kc.Brokers(ctx); наружу выставлен порт 9093, а не 9092 - топик — CreateTopics с одной партицией и репликацией 1 или авто-создание брокером - читатель без группы читает нулевую партицию с FirstOffset; дедлайн на ReadMessage обязателен - CleanupContainer в тесте или TerminateContainer в defer — уборка при любом исходе ### Проверка: старый тег образа **Вопрос:** kafka.Run вызвали с образом confluent-local:5.5.0. Что произойдёт? - Контейнер поднимется: модуль не проверяет теги образов - ✅ Run вернёт ошибку: KRaft в этом теге ещё не появился, а модуль работает только на нём - Контейнер стартует, а первый тест упадёт с отказом соединения - Модуль молча подставит совместимый тег вместо указанного Для официального образа модуль сверяет тег до старта контейнера: версии ниже 7.4.0 отвергаются ошибкой про KRaft-режим — Docker даже не потревожен. Чужие образы без проверки пропускаются: стартует их содержимое, и гарантий конфигурации модуль на них не даёт. ### Что выведет Brokers ```go kc, err := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0", kafka.WithClusterID("test-cluster"), ) // err == nil brokers, err := kc.Brokers(ctx) fmt.Println(brokers[0]) ``` **Что выведет программа?** - ✅ localhost:<маппированный порт> — хост и порт, выданный Docker - localhost:9092 — стандартный порт брокера Kafka - PLAINTEXT://localhost:<маппированный порт> — со схемой листенера - Ошибка: брокер ещё не готов принимать клиентов Brokers собирает адрес через PortEndpoint по порту 9093/tcp: хост localhost, порт — маппированный, он меняется от запуска к запуску. Схему метод не дописывает: строка подходит и для kafka.TCP у Writer, и для списка адресов клиента. Готовность к этому моменту уже проверена — Run вернулся после строки о переходе брокера в RUNNING. ### Соберите запуск брокера Восстановите вызовы: старт брокера, уборка, адреса. ```go kc, err := kafka.___(ctx, "confluentinc/confluent-local:7.5.0") testcontainers.___(t, kc) brokers, err := kc.___(ctx) ``` Подсказка: Брокер стартует kafka.Run, за уборку в тесте отвечает CleanupContainer, адреса отдаёт метод Brokers. Правильные ответы в пропусках: `Run`, `CleanupContainer`, `Brokers` ### Расставьте путь сообщения в тесте Соберите путь: контекст, запуск брокера, уборка, адрес, продюсер, отправка, чтение, проверка. Правильный порядок строк: 1. `ctx := context.Background()` 2. `kc, err := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0")` 3. `testcontainers.CleanupContainer(t, kc)` 4. `brokers, err := kc.Brokers(ctx)` 5. `writer := &kafka.Writer{Addr: kafka.TCP(brokers...), Topic: "orders"}` 6. `err = writer.WriteMessages(ctx, kafka.Message{Value: []byte("paid")})` 7. `msg, err := reader.ReadMessage(readCtx)` 8. `if string(msg.Value) != "paid" { t.Fatal("сообщение не дошло") }` ### Карточки: модуль Kafka - **kafka.Run** — Вход в модуль: образ обязателен и идёт первым аргументом, дальше кастомайзеры. Поднимает один брокер в KRaft-режиме и возвращает типизированный KafkaContainer. - **Brokers(ctx)** — Адреса брокеров: срез с единственным элементом — PortEndpoint по порту 9093/tcp. Опции про число брокеров у модуля нет. - **WithClusterID** — Пишет переменную CLUSTER_ID в окружение контейнера; после старта её значение попадает в поле ClusterID. - **KRaft** — Режим без Zookeeper: роли broker и controller в одном процессе, хранилище форматирует kafka-storage. Официальный образ ниже 7.4.0 модуль отвергает. - **Reader без GroupID** — Чтение одной партиции — нулевой по умолчанию — без коммитов; оффсет нового читателя стартует с FirstOffset. Validate запрещает GroupID вместе с Partition. - **9093 против 9092** — 9093 — PLAINTEXT-листенер для клиентов, он маппируется наружу; 9092 — внутренний BROKER-листенер; 9094 — контроллер. Brokers возвращает адрес 9093. > **Про песочницу: нужен локальный Docker.** Честно: в браузерной песочнице этот урок не запустить — Docker там нет, а брокеру нужен живой демон. Рабочая схема — локально: Go, запущенный Docker и обычный go test в терминале. Код разобран построчно, поэтому напечатайте тест руками и прогоните у себя: первый запуск скачает образ confluent-local — он тяжелее postgres из прошлых уроков, — зато дальше образ кешируется и старты ускоряются. Если тест висит на чтении, посмотрите docker logs контейнера: строка о переходе в RUNNING скажет, дошёл ли брокер до готовности. Что дальше: брокер — не последний сетевой модуль курса. Следующий урок — [модуль RabbitMQ: обменники и очереди](/courses/testcontainers/lessons/09-rabbitmq-module/) — покажет ту же схему Run плюс метод-адрес для очередей с обменниками и маршрутами. Для Kafka-сервисов углубляйтесь в клиент: [курс kafka-go](/courses/kafka-go/) ведёт от первого Writer до групп, ретраев и админ-API. Обзор всего пути — [курс testcontainers](/courses/testcontainers/): от wait-стратегий до CI и параллельности. И упражнение на честность перед следующей темой: создайте топик с тремя партициями, отправьте несколько сообщений с разными ключами и прочитайте его читателем без группы — пропажа «лишних» сообщений объяснит привязку читателя к нулевой партиции лучше любых слов. ## Коротко: что запомнить - kafka.Run поднимает один брокер в KRaft-режиме без Zookeeper: образ — первый аргумент, WithClusterID задаёт CLUSTER_ID, он же попадает в поле ClusterID. - Brokers(ctx) возвращает срез с одним адресом: наружу выставлен порт 9093, внутренние листенеры 9092 и 9094 не маппируются. - Топик создаёт админ-API клиента или сам брокер при первой отправке; на одиночном брокере фактор репликации — единица. - Читатель без группы читает нулевую партицию с FirstOffset; ReadMessage требует дедлайна, иначе тест висит до таймаута пакета. ## Домашнее задание Соберите пакет с TestMain: kafka.Run с WithClusterID, TerminateContainer в defer и выводом Brokers в лог. Затем напишите тест «сообщение дошло»: CreateTopics с одной партицией и репликацией 1, читатель без группы, продюсер с ключом, ReadMessage с дедлайном в десять секунд. После этого создайте второй топик с тремя партициями, отправьте туда несколько сообщений с разными ключами и прочитайте его тем же читателем без группы — объясните по устройству партиций, почему вы увидели только часть отправленного. В конце уберите дедлайн с ReadMessage, отправьте сообщение в топик, который никто не читает, и наблюдайте зависший тест; верните дедлайн и убедитесь, что тест краснеет сам. ## Вопросы и ответы по уроку ### Почему Brokers возвращает срез, а не одну строку? Формат под общий случай: адресов у Kafka столько, сколько брокеров, и клиентские библиотеки принимают список. Модуль поднимает одиночный брокер, поэтому в срезе один элемент, но сигнатура не изменится, если вы соберёте кластер сами из нескольких контейнеров в общей сети. Клиенты разматывают срез напрямую: kafka.TCP принимает адреса через многоточие и превращает их в адрес кластера, так что переход от одного брокера к нескольким не потребует править вызывающий код. В тестах удобно опираться на Brokers, а не на самодельный разбор ответа инспекции: срез адресов — контракт модуля, и он стабильнее внутренних полей контейнера. ### Можно ли поднять несколько брокеров через модуль? Опции про число брокеров у Run нет: модуль настраивает ровно один — единственный номер брокера, факторы репликации в единицу, advertised listeners на один адрес. Кластер собирают из нескольких контейнеров модуля в общей Docker-сети: модуль расставляет адреса кворума контроллера по первому алиасу сети, если сеть задана. Для большинства тестов этого не требуется: один брокер проверяет контракты продюсера и консьюмера, а отказоустойчивость интеграционно проверяют редко. ### Сообщение отправлено, а ReadMessage его не видит. С чего начать? С трёх мест. Первое — партиции: читатель без группы смотрит в нулевую, и сообщения с другими ключами могли уехать в соседние партиции. Второе — имена топиков у продюсера и читателя: опечатка здесь не даёт ошибок, просто работают два разных топика. Третье — дедлайн: без таймаута на контексте тест не падает, а висит до общего лимита пакета и выглядит как зависание. Логи контейнера покажут, дошёл ли брокер до строки о переходе в RUNNING.