Модуль Kafka: продюсер и консьюмер

Брокер сообщений — второй по частоте гость интеграционных тестов после базы: заказы, уведомления, события между сервисами. Заменить его моком просто, но тогда тест не проверяет почти ничего из того, ради чего 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)

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 разбирает Writer, Reader и админ-клиент по косточкам, здесь нам хватит трёх вызовов.

Тест «сообщение дошло»

// 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. Что вернёт метод?

Задача → приём: 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: больше брокер не прокормит. Несколько партиций ломают порядок: ключи раскладываются по партициям, и два сообщения одного сценария приходят читателю в порядке внутри партиции, а не внутри сценария. Читатель без группы привязан к нулевой партиции — при нескольких он увидит её одну и решит, что сообщения потерялись. Третья — много брокеров: модуль их не собирает, это несколько контейнеров в общей сети, тема урока про сети.

Чек-лист 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. Что произойдёт?

Что выведет Brokers

Прочитайте код и выберите, что он выведет:

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])

Соберите запуск брокера

Восстановите вызовы: старт брокера, уборка, адреса.

kc, err := kafka.(ctx, "confluentinc/confluent-local:7.5.0") testcontainers.(t, kc) brokers, err := kc.(ctx)

Расставьте путь сообщения в тесте

Соберите путь: контекст, запуск брокера, уборка, адрес, продюсер, отправка, чтение, проверка.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: модуль Kafka

Термин

kafka.Run

нажмите, чтобы перевернуть

Определение

Вход в модуль: образ обязателен и идёт первым аргументом, дальше кастомайзеры. Поднимает один брокер в KRaft-режиме и возвращает типизированный KafkaContainer.

1 / 6

Про песочницу: нужен локальный Docker

Честно: в браузерной песочнице этот урок не запустить — Docker там нет, а брокеру нужен живой демон. Рабочая схема — локально: Go, запущенный Docker и обычный go test в терминале. Код разобран построчно, поэтому напечатайте тест руками и прогоните у себя: первый запуск скачает образ confluent-local — он тяжелее postgres из прошлых уроков, — зато дальше образ кешируется и старты ускоряются. Если тест висит на чтении, посмотрите docker logs контейнера: строка о переходе в RUNNING скажет, дошёл ли брокер до готовности.

Что дальше: брокер — не последний сетевой модуль курса. Следующий урок — модуль RabbitMQ: обменники и очереди — покажет ту же схему Run плюс метод-адрес для очередей с обменниками и маршрутами. Для Kafka-сервисов углубляйтесь в клиент: курс kafka-go ведёт от первого Writer до групп, ретраев и админ-API. Обзор всего пути — курс testcontainers: от wait-стратегий до CI и параллельности. И упражнение на честность перед следующей темой: создайте топик с тремя партициями, отправьте несколько сообщений с разными ключами и прочитайте его читателем без группы — пропажа «лишних» сообщений объяснит привязку читателя к нулевой партиции лучше любых слов.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

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.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.