Admin API: топики из кода

kafka.Client — маленькая структура из трёх полей, и конструктора у неё нет: клиент создаётся литералом. Addr — адрес кластера или конкретного брокера типа net.Addr, его даёт хелпер kafka.TCP("localhost:9092"), знакомый по третьему уроку. Timeout — лимит на каждый административный запрос; ноль означает, что таймаута нет, и это плохой дефолт для продакшена. Transport — интерфейс RoundTripper, через который уходят запросы; при nil библиотека подставит DefaultTransport, но поле открывает дорогу общему транспорту с таймаутами и своим логгером для всего сервиса. Клиент не держит постоянных соединений: каждый вызов — отдельный round trip, поэтому создание одного объекта на весь процесс безопасно и дёшево. Важно понимать масштаб: Writer и Reader отвечают за данные, Client — за управление кластером, и последний ничего не знает про продюсирование. В третьем уроке литерал уже мелькал для коммитов оффсетов; здесь он становится полноправным инструментом. Одна деталь экономит час отладки: Timeout в клиенте не совпадает с таймаутами Writer и Reader — это отдельный лимит административных запросов, и он настраивается независимо, под свой SLA на управление кластером.

CreateTopics принимает CreateTopicsRequest с адресом и списком TopicConfig — одним вызовом можно создать сразу несколько топиков. В TopicConfig четыре рабочих поля: Topic — имя, NumPartitions — число партиций, ReplicationFactor — количество копий каждой партиции, ConfigEntries — брокерские настройки топика. Каждая настройка — это kafka.ConfigEntry с двумя строковыми полями, ConfigName и ConfigValue: протокол передаёт конфиги парами имя-значение, поэтому retention.ms записывается как "86400000", а min.insync.replicas из пятнадцатого урока — как "2". Есть и поле ValidateOnly: при true брокер проверит конфигурацию и не создаст ничего — сухой прогон для CI, который отчитывается, а не меняет кластер. Номер партиции, выбранной по ключу, зависит от NumPartitions, так что решение о числе партиций здесь всплывает снова — как в пятом уроке, только теперь вы его принимаете кодом. Ответ CreateTopicsResponse несёт карту Errors: имя топика указывает на error, и у каждого имени свой вердикт. Транспортная ошибка err означает, что запрос не состоялся вовсе, а вердикты внутри карты сравниваются через errors.Is с кодами пакета: TopicAlreadyExists, InvalidReplicationFactor, InvalidConfiguration.

Сервис, который создаёт свои топики при каждом старте, рано или поздно запустится на окружении, где топик уже есть. Брокер ответит кодом TopicAlreadyExists — «a topic with this name already exists» — и наивная проверка if err != nil превратит второй деплой в аварию. Правильная трактовка: TopicAlreadyExists при создании — не сбой, а штатный исход, как nil, и для этого идемпотентная обёртка ensureTopic стала ритуалом при старте Go-сервисов. Два правила функции: создание идемпотентно, но конфигурацию оно не гарантирует, и проверка существования без сверки параметров — половина решения. Но у неё есть слепая зона, о которой стоит помнить: CreateTopics создаёт, но не обновляет. Если топик существует с одной партицией, а в запросе просят шесть, брокер вернёт тот же TopicAlreadyExists и оставит конфиг прежним — число партиций наращивается отдельным методом CreatePartitions, а конфиги существующего топика правятся через AlterConfigs. Поэтому честный ensureTopic после «уже существует» сверяет фактическую конфигурацию с желаемой через Metadata, иначе расхождение между ожиданиями кода и реальностью кластера обнаружится в продакшене, а не на деплое.

Список топиков и их устройство в kafka-go достаются одним методом — client.Metadata. Запрос MetadataRequest несёт адрес и список имён; пустой список означает «весь кластер», конкретные имена — «опишите только их». Ответ MetadataResponse богаче CLI-вывода: ClusterID, контроллер кластера, список брокеров с адресами и идентификаторами, и главное — Topics, где у каждого топика есть имя, флаг Internal и массив партиций с лидерами и репликами. Флаг Internal помечает системные топики вроде __consumer_offsets, которые CLI иногда прячет, а вы в собственных скриптах должны фильтровать сами. Формально это один вызов на роль двух команд консоли: list topics и describe topics в нашей терминологии — ListTopics и DescribeTopics — собраны в Metadata. Ответ читается как карта: имя топика — ключ, а значение — его партиции с лидерами и репликами, и по ней удобно сверять фактический расклад с ожидаемым перед стартом группы консьюмеров из седьмого урока. В библиотеке есть и экспериментальный пакет topics с функцией List поверх него, но без обещаний обратной совместимости, поэтому прямые вызовы Metadata надёжнее для продакшена.

DeleteTopics — самая короткая и самая опасная ручка клиента: DeleteTopicsRequest со списком имён, ответ с той же картой Errors, и топики с их данными исчезают без корзины — ни подтверждений, ни восстановления. Удаление асинхронно размечает партиции, и брокер ещё несколько секунд может отвечать на Metadata именем, которого больше нет. Из этой асимметрии рождается организационное правило урока. В разработке и CI код создаёт и чинит окружение сам: песочница поднимается, ensureTopic создаёт топики, тесты гоняются, всё удаляется одной командой — это легитимная автоматизация, ради которой admin API и писался. В продакшене топик — это бюджет: ReplicationFactor умножает хранилище, число партиций держит лимиты файловых дескрипторов брокеров, retention определяет срок жизни данных. Эти решения принимают люди или инфраструктура как код, а сервис при старте проверяет существование и падает с внятной ошибкой, если топика нет. DeleteTopics из кода сервиса в проде — красный флаг код-ревью почти всегда. Даже уборка тестовых окружений должна проходить через отдельный скрипт с явным списком имён, а не через общий конфиг сервиса.