Тесты: изоляция по топикам

Проверить kafka-код сложнее, чем обычные функции на входе и выходе: Writer и Reader — внешние зависимости с сетью, состоянием брокера и временем ответа. Наивный путь — завести один общий тестовый топик на локальном брокере и слать туда всё подряд; такой набор тестов живёт до первого параллельного прогона, когда два пакета начинают читать чужие сообщения. Отсюда развилка, которую проходит каждый, кто тестирует kafka-код. Первая стратегия — интеграционная: настоящий брокер в Docker, тест создаёт себе топик, гоняет сценарий, сверяет результат и убирает за собой. Она проверяет всё: сериализацию, ключи и партиции, ретраи Writer, коммиты Reader — но стоит секунд на прогон и требует окружения. Вторая — юнитовая: код сервиса развязывается с конкретным типом kafka.Writer через собственный интерфейс, и вместо брокера в тестах работает стаб. Она мгновенна и детерминированна, но не видит сеть и протокол, поэтому сам клиент проверить не способна. Зрелый набор тестов использует обе: стабы — на бизнес-правила, брокер — на интеграцию с kafka-go. Продюсер собирали в четвёртом уроке, Reader разбирали в четырнадцатом — теперь научимся проверять оба слоя так, чтобы тесты не ссорились друг с другом и не врали.

Интеграционные тесты: брокер в Docker#

Интеграционный тест kafka-кода начинается не с Writer, а с окружения: нужен живой брокер, адрес которого тест узнаёт снаружи. Классический стенд — Docker: compose-файл с одним узлом Kafka для быстрых прогонов или testcontainers, поднимающий брокер на свободном порту прямо из теста. В обоих случаях адрес приходит через переменную окружения или флаг, и тест либо работает, либо честно пропускается через t.Skip, когда брокера нет: красная сборка из-за отсутствующей среды вредит больше, чем пропущенный тест. Дальше — изоляция данных. Один брокер обслуживает всех, но каждый тест обязан жить в собственном топике: создание через Client и CreateTopics занимает у брокера миллисекунды, а экономит часы отладки. Это Admin API из шестнадцатого урока: клиент с адресом брокера, запрос CreateTopicsRequest с TopicConfig — и топик готов с нужным числом партиций. Топик, созданный тестом, принадлежит тесту: в нём нет чужих сообщений, его офсеты начинаются с нуля, и Reader прочитает ровно то, что записал сценарий. По окончании тест удаляет топик через DeleteTopics — или поручает это t.Cleanup, который сработает даже после t.Fatal посреди сценария.

Тест с уникальным топиком и очисткой

func TestOrderProducer(t *testing.T) {
        ctx := context.Background()
        broker := os.Getenv("KAFKA_BROKER")
        topic := fmt.Sprintf("orders-test-%d", time.Now().UnixNano())

        client := &kafka.Client{Addr: kafka.TCP(broker)}
        _, err := client.CreateTopics(ctx, &kafka.CreateTopicsRequest{
                Topics: []kafka.TopicConfig{{
                        Topic:         topic,
                        NumPartitions: 1,
                }},
        })
        if err != nil {
                t.Fatalf("create topic: %v", err)
        }
        t.Cleanup(func() {
                _, _ = client.DeleteTopics(ctx, &kafka.DeleteTopicsRequest{
                        Topics: []string{topic},
                })
        })

        writer := &kafka.Writer{
                Addr:  kafka.TCP(broker),
                Topic: topic,
        }
        defer writer.Close()
        // далее: WriteMessages и сверка содержимого через Reader
}

Имя топика собирает fmt.Sprintf с UnixNano: двух прогонов с одинаковым названием не бывает даже при параллельном запуске пакетов. CreateTopics из Admin API создаёт топик явно — автосоздание брокера отдало бы конфигурацию на откуп дефолтам. t.Cleanup регистрирует DeleteTopics сразу после создания и срабатывает при любом исходе теста, включая t.Fatalf: отложенные вызовы выполняются всегда, когда тест завершается.

Проверка: t.Cleanup

Тест создал топик, зарегистрировал t.Cleanup с DeleteTopics, а посреди сценария упал через t.Fatalf. Что случится с топиком?

Живое место: тесты в одном топике

Инцидент на CI: пакет тестов гонялся параллельно против общего топика orders-test, и интеграционная проверка каждые несколько сборок ловила чужое сообщение — Reader вместо ожидаемого события получал payload соседнего сценария и падал с расхождением. Отладить было больно: сбой воспроизводился раз в день, при совпадении таймингов двух джобов, и не повторялся локально. Лечение заняло вечер: каждый тест стал собирать имя топика через fmt.Sprintf с UnixNano, создавать топик сам и убирать за собой через t.Cleanup. С тех пор красный тест означает баг в коде, а не unlucky-расписание планировщика. Изоляция по топикам — дешёвая страховка: брокер многопользовательский по устройству, аренда топика на прогон ничего не стоит, а расследование плавающих падений стоит дорого.

Уникальный топик на каждый прогон#

Механика изоляции проста, но у неё есть детали, которые отличают аккуратный набор тестов от хрупкого. Имя топика строится из префикса сценария и уникального суффикса: fmt.Sprintf("payments-retry-%d", time.Now().UnixNano()) — или случайная строка из math/rand. Префикс по имени сценария — не мелочь: без него забытый топик в общей среде анонимен. Уникальность нужна не для красоты, а для честности параллельности: go test с флагом -parallel, разные пакеты, разные джобы CI — каждый прогон получает собственный лог сообщений, и брокер физически не может скрестить данные, потому что топики не пересекаются. Вторая деталь — создавать топик самостоятельно, а не надеяться на автосоздание брокера: auto.create.topics.enable в прод-конфигурации часто выключен, а созданный брокером топик наследует дефолтные число партиций и retention, которые тест не контролирует. Явный CreateTopics из шестнадцатого урока фиксирует и имя, и конфигурацию: NumPartitions для детерминированного порядка, ReplicationFactor под стенд. Третья — очистка: DeleteTopics в t.Cleanup держит брокер чистым даже после падений, и это вежливость к коллегам по общей тестовой среде, а не только вопрос места на диске локального Docker.

Что увидит Reader без уникального суффикса?

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

// Два теста идут параллельно, каждый пишет
// в топик "orders-test" и ждёт одно своё сообщение:
// тест A шлёт key=order-42, тест B шлёт key=order-43.
// Reader в каждом тесте читает первую запись топика.
m, err := reader.FetchMessage(ctx)

Интерфейс Producer вместо конкретного Writer#

kafka.Writer — конкретный тип, структура из writer.go, и код, который принимает её напрямую, невозможно проверить без брокера: подменить нечем. Лекарство — тонкий интерфейс на стороне потребителя: Producer с одним методом Write, принимающим контекст, ключ и значение. Продовая реализация оборачивает Writer и транслирует вызов в WriteMessages со всеми ретраями и разбором WriteErrors из двенадцатого урока; тестовая — стаб. Интерфейс определяет тот, кто им пользуется, а не библиотека: в нём ровно те операции, которые нужны сервису, и ни одного поля Balancer или RequiredAcks — это детали kafka-go, чужие бизнес-коду. Для консюмера symmetric ход: собственный интерфейс источника сообщений с FetchMessage и Commit, поверх которого работает обёртка над kafka.Reader из четырнадцатого урока. Цена решения — один слой индирекции и десяток строк обёртки; выигрыш — бизнес-логика тестируется мгновенно и детерминированно, а специфика kafka-go остаётся в паре интеграционных тестов самой обёртки. Батч-поведение Writer из десятого урока проверяется отдельно и только против брокера: стаб ничего не знает о бэкоффе и пачках.

Интерфейс и продовая обёртка

type Producer interface {
        Write(ctx context.Context, key string, value []byte) error
}

type KafkaProducer struct {
        writer *kafka.Writer
}

func NewKafkaProducer(w *kafka.Writer) *KafkaProducer {
        return &KafkaProducer{writer: w}
}

func (p *KafkaProducer) Write(ctx context.Context, key string, value []byte) error {
        return p.writer.WriteMessages(ctx, kafka.Message{
                Key:   []byte(key),
                Value: value,
        })
}

Интерфейс объявлен рядом с потребителем, а не в пакете обёртки — так принято в Go: небольшие интерфейсы определяет тот, кто их вызывает. KafkaProducer хранит конкретный *kafka.Writer и служит единственным местом, где сервис касается kafka-go. Ошибка WriteMessages проходит наружу без пересборки: разбор WriteErrors — задача вызывающего кода, как в двенадцатом уроке.

Проверка: чей интерфейс

Почему интерфейс Producer объявлен в коде сервиса, а не экспортирован из библиотеки-обёртки над kafka-go?

Стабы и проверка содержимого#

Стаб Producer — несколько строк, которые превращают тест бизнес-логики в мгновенный: он копит сообщения в слайс и по желанию возвращает заготовленную ошибку. Тест проверяет, какие ключи и значения ушло из сервиса, в каком порядке, и что происходит при отказе записи: заказ не потерян, ошибка залогирована, ретрай запланирован. Для консюмера зеркальный стаб источника: тест подаёт сообщения в канал и смотрит, как обработчик переживает дубликаты, битые payload и пустые ключи. Стабы не видят сеть — и это их сила: детерминированность дороже правдоподобия в юнит-контуре. Зато интеграционная проверка содержимого замыкает круг против настоящего брокера: после WriteMessages тестовый Reader открывает тот же топик, читает FetchMessage и сверяет Key и Value с ожидаемыми, затем CommitMessages. Это единственный способ убедиться, что дошло именно то, что отправляли: сериализация не исказила payload, ключ попал в ожидаемую партицию, Writer не съел сообщение молча. Такая сверка — сердце интеграционного теста продюсера, и делается она тем же kafka.Reader, который обслуживает продовый консюмер, а не отдельной утилитой.

Стаб Producer и сверка через Reader

type StubProducer struct {
        Written []kafka.Message
}

func (s *StubProducer) Write(_ context.Context, key string, value []byte) error {
        s.Written = append(s.Written, kafka.Message{
                Key:   []byte(key),
                Value: value,
        })
        return nil
}

// Интеграционная сверка: читаем топик обратно.
reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: brokers,
        Topic:   topic,
        GroupID: "verify",
})
defer reader.Close()

m, err := reader.FetchMessage(ctx)
if err != nil {
        t.Fatalf("fetch: %v", err)
}
if string(m.Key) != "order-42" {
        t.Fatalf("key = %q, want order-42", m.Key)
}
reader.CommitMessages(ctx, m)

StubProducer реализует тот же интерфейс Producer: сервис не замечает подмены, а тест получает слайс записанного для сверки. Во второй половине — круговая проверка против брокера: тестовый Reader с GroupID читает топик FetchMessage, сравнивает ключ и коммитит. Это тот же API продового консюмера, поэтому сверка проверяет и сериализацию, и партиционирование.

Про песочницу: где идут интерактивы

Интерактивы этого урока разыгрывают тестовые сценарии без сети: брокер не нужен ни квизам, ни пазлам. Реальные прогоны делайте локально: Docker с одиночным брокером поднимается за минуту, адрес передаётся через KAFKA_BROKER, а t.Skip пропустит интеграционные тесты, если контейнер не запущен. Интеграционные тесты честно медленнее юнитов — секунды против миллисекунд, поэтому держите их в отдельном пакете с билд-тегом и не смешивайте с быстрым контуром, который запускается на каждом сохранении. В песочнице курса сеть закрыта: брокерный слой закрепляется квизами, живой прогон остаётся за вашей машиной.

Гигиена тестового окружения#

Адрес брокера — из окружения или флага: KAFKA_BROKER, никаких захардкоженных адресов в коде теста.
Нет брокера — t.Skip с объяснением: красная сборка от отсутствующей среды вредит больше, чем пропущенный тест.
Уникальное имя топика на прогон: fmt.Sprintf с UnixNano или случайным суффиксом.
Топик создаёт сам тест через CreateTopics и удаляет через t.Cleanup с DeleteTopics.
testcontainers поднимает брокер на свободном порту: фиксированные порты конфликтуют в общей среде.
Таймаут контекста на тест: зависший FetchMessage не должен вешать сборку до общего лимита CI.

Гигиена: адреса, порты и окружение#

Последний слой — окружение, на котором сыпятся даже правильно изолированные тесты. Захардкоженный адрес localhost:9092 работает на ноутбуке и падает в CI, где брокер слушает другой порт или живёт в соседнем контейнере под своим именем; адрес должен приходить из переменной окружения или флага, а testcontainers вообще снимает вопрос, поднимая брокер на свободном порту к началу теста. Вторая привычка — таймауты: контекст с дедлайном вокруг FetchMessage и WriteMessages, чтобы зависший брокер превращался в красный тест через секунды, а не в повисшую сборку до общего лимита джобы; механика дедлайнов — тема восемнадцатого урока. Третья — честный Skip: если KAFKA_BROKER пуст, интеграционный тест сообщает причину и выходит, не создавая ложных падений. Собранные вместе, эти правила дают набор тестов, который детерминирован при параллельных прогонах, чистит за собой и одинаково работает на ноутбуке и в CI. Таблица ниже сводит выбор инструмента: что проверяем, чем и на каком уровне.

Что тестируем и чем#

Что проверяем	Инструмент	Уровень
Бизнес-логику продюсера: ключи, значения, ветвления	Стаб Producer со слайсом записанного	Юнит, без брокера
Реакцию сервиса на отказ записи	Стаб с заготовленной ошибкой	Юнит, без брокера
Обработку сообщений консюмером: дубликаты, битые payload	Стаб источника сообщений	Юнит, без брокера
Реальные WriteMessages против брокера	Writer и уникальный топик в Docker	Интеграция
Содержимое топика: сериализация, ключи, партиции	Тестовый Reader, FetchMessage и сверка	Интеграция

Соберите хелпер тестового топика

Допишите вызов форматирования уникального имени и регистрацию очистки в хелпере, который создаёт топик для теста.

func newTestTopic(t *testing.T, broker string) string { t.Helper() topic := .Sprintf("orders-test-%d", time.Now().UnixNano()) client := &kafka.Client{Addr: kafka.TCP(broker)} // ... CreateTopics для topic ... t.(func() { _, _ = client.DeleteTopics(ctx, &kafka.DeleteTopicsRequest{ Topics: []string{topic}, }) }) return topic }

Соберите хелпер по шагам

Расставьте строки хелпера newTestTopic: сигнатура с *testing.T, Helper, уникальное имя, клиент, создание топика, обработка ошибки, очистка, возврат имени.

Строки

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

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

Карточки: изоляция и стабы

Термин

Уникальный топик на прогон

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

Определение

fmt.Sprintf с UnixNano или случайным суффиксом: параллельные тесты не читают чужие сообщения и не спорят за офсеты.

1 / 6

Две стратегии сошлись в одну систему: бизнес-код зависит от собственного интерфейса и проверяется стабами мгновенно, а единственная обёртка над Writer выезжает на интеграцию с уникальным топиком, тестовым Reader и t.Cleanup. Дальше по курсу: урок про Admin API раскроет CreateTopics и DeleteTopics в полный рост — управление топиками, конфигами и партициями; урок про батчи покажет, что в Writer проверяется только против живого брокера и почему стаб здесь бессилен; урок про ошибки продюсера даст разбор WriteErrors, который стоит повторить обёртке; а урок про первый продюсер напомнит, с чего начинался Writer, который теперь тестируем. Для живого прогона достаточно локального Docker-брокера из третьего урока: поднимите контейнер, передайте адрес через KAFKA_BROKER и запустите интеграционный пакет — изоляция по топикам позволит гонять его параллельно хоть весь вечер.

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

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

Вывод

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

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

Две стратегии тестов kafka-кода: интеграция с реальным брокером и юниты на стабах за собственными интерфейсами.
Уникальный топик на прогон: fmt.Sprintf с UnixNano, изоляция параллельных прогонов гарантирована непересекающимися логами.
Тест создаёт топик через CreateTopics и удаляет через t.Cleanup с DeleteTopics: очистка срабатывает даже после t.Fatal.
kafka.Writer — конкретный тип; для подмены сервису нужен свой интерфейс Producer с методом Write.
Стаб Producer копит сообщения в слайс и возвращает заготовленную ошибку: проверяются ключи, порядок и реакция на отказ.
Проверка содержимого: тестовый Reader читает топик FetchMessage, сверяет Key и Value, затем CommitMessages.
Гигиена: адрес брокера из окружения, t.Skip без брокера, таймауты контекста, testcontainers для свободных портов.
Стабы — для бизнес-правил, брокер — для интеграции с kafka-go: батчи, ретраи и партиционирование стаб не проверяет.

Домашнее задание

Соберите хелпер newTestTopic: уникальное имя через fmt.Sprintf с UnixNano, CreateTopics через Client, DeleteTopics в t.Cleanup — и напишите на нём интеграционный тест продюсера, который после WriteMessages читает топик тестовым Reader и сверяет ключ и значение. Затем введите в сервисе интерфейс Producer, продовую обёртку KafkaProducer и стаб: перепишите бизнес-тесты на стабы, оставив обёртке один интеграционный тест. Проверьте гигиену: адрес брокера приходит из KAFKA_BROKER, при пустой переменной тест уходит в t.Skip с причиной, контекст ограничен таймаутом. Финал: запустите пакет дважды параллельно и объясните по устройству топиков, почему прогоны не мешают друг другу.

Вопросы и ответы по уроку

Можно ли тестировать kafka-код против эмулятора брокера в памяти вместо Docker?

Эмуляторы существуют, но они имитируют протокол, а не поведение брокера: ретраи, ребалансировку, сжатие ISR и eviction офсетов они не воспроизводят, и тест зелёный там, где прод падает. Практичная замена — две стратегии урока: стабы на бизнес-правила и настоящий брокер в Docker на интеграцию. Одиночный узел закрывает механику Writer и Reader; кластерные сценарии — отдельная история, которой хватит стенда из трёх нод.

Почему не принять *kafka.Writer напрямую и не стабить библиотечный тип?

kafka.Writer — конкретная структура, а не интерфейс: подменить её реализацию в чужом пакете невозможно, компилятор потребует именно этот тип. Поэтому контракт объявляет потребитель: один метод Write с ровно теми аргументами, которые нужны сервису. Такой интерфейс стабится тремя строками, не тянет поля конфигурации и переживает смену библиотеки: если обёртка когда-нибудь сменится, бизнес-код и его тесты не заметят.

Как тестировать порядок сообщений, если Writer шлёт параллельно?

Порядок в Kafka гарантирован только внутри партиции, поэтому тест порядка делают детерминированным: топик с NumPartitions: 1 и записи с пустым или одинаковым ключом ложатся строго по очереди, и Reader прочитает их в порядке отправки. Партиционирование по ключам из пятого урока проверяют иначе: создают топик с несколькими партициями и сверяют, что один ключ всегда попадает в одну партицию, не требуя конкретного номера. Стабы здесь не помогут: раскладка по партициям — работа брокера и балансера.

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

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