Первый продюсер: Writer

Writer создаётся функцией kafka.NewWriter, которой передаётся kafka.WriterConfig — и вот первый сюрприз: NewWriter ничего не подключает. Конструктор лишь запоминает настройки; соединения с брокерами появятся при первой записи, а Writer будет поддерживать их пул дальше. Сердце конфига — Brokers, список адресов брокеров, и Topic — топик по умолчанию, куда уйдут все сообщения этого Writer. Поле Balancer решает, как сообщения распределяются по партициям топика; если его не задать, работает RoundRobin — по кругу. Батчевые настройки BatchSize и BatchTimeout влияют на производительность и подробно разобраны в десятом уроке, здесь оставим их по умолчанию. Важно правило проектирования: Writer — долгоживущий объект уровня приложения. Он потокобезопасен, держит внутреннюю очередь и пул соединений, и создавать его на каждое сообщение — самый известный антипаттерн kafka producer на golang: каждый конструктор — это новые соединения и новое ожидание, а выигрыша ноль. Собрали один Writer в main, передали в обработчики — и живёте с ним весь процесс. Список адресов в Brokers нужен не для красоты: Writer узнаёт у первого живого брокера метаданные кластера — где какой лидер партиций — и дальше общается уже напрямую с ними. Если захочется альтернативного синтаксиса — есть литерал kafka.Writer с полем Addr: библиотека поддерживает оба пути, курс держится классического NewWriter c WriterConfig.

Внутри WriteMessages скрыта конвейерная работа, и о ней полезно знать до первого инцидента. Вызов складывает сообщение во внутреннюю очередь Writer, откуда отдельные горутины собирают их в батчи по адресату — топик и партиция определяются балансировщиком. Батч уезжает по сети на лидера партиции, тот дописывает данные в свой лог и отвечает подтверждением; уровень ожидания задаётся RequiredAcks: конструктор NewWriter при нулевом значении подставляет kafka.RequireAll — запись считается состоявшейся, когда её подтвердили все реплики из ISR; на локальном брокере с одной репликой это просто подтверждение лидера, а сниженный порог kafka.RequireOne и его цена — тема урока про durability. Только после ответа брокера WriteMessages возвращает управление — поэтому «отправил» в этом коде означает «брокер подтвердил запись», а не «байты ушли в сокет». Для вызывающего кода это удобный контракт: при успехе сообщение уже в логе, и о нём можно забыть до чтения. Обратная сторона — запись синхронная по умолчанию: редкие события это устраивает, а для высоких скоростей есть асинхронный режим и батчи — оба разберём в десятом уроке. Кстати, именно поэтому продюсер здесь называется Writer: библиотека подчёркивает метафору дописывания в журнал — вы не шлёте письмо в никуда, а вносите запись в лог, который переживёт и ваш процесс.

Сигнатура WriteMessages принимает контекст и переменный список сообщений: WriteMessages(ctx, msg) для одиночной отправки и WriteMessages(ctx, msgs...) для пачки, собранной в срез. Сообщение — это kafka.Message с полями Key и Value типа []byte: Kafka не знает про JSON, protobuf и ваши структуры, она хранит байты, и сериализация — тема одиннадцатого урока. Key внешне похож на опцию, но на самом деле это маршрутизатор: одинаковый ключ всегда попадает в одну партицию, что и создаёт порядок событий по сущности — заказу, аккаунту, пользователю. Если WriterConfig.Topic пуст, топик можно указать в каждом сообщении через поле Message.Topic — так один Writer обслуживает несколько каналов. Пачка за один вызов экономит не только строки кода, но и сетевые поездки: одно сообщение — один produce-запрос, тысяча — один-два в зависимости от размера. Наконец, WriteMessages безопасно звать из нескольких горутин одновременно: очередь Writer сериализует работу, и мьютексы в вызывающем коде не нужны. Отдельно про размер: Value не обязан быть маленьким, но у брокера есть потолок на сообщение — по умолчанию около мегабайта, поэтому большие файлы в топик не таскают: их место в объектном хранилище, а в сообщении — ссылка.

Честная обработка ошибок начинается с понимания, откуда они берутся. Первая группа — недоступный брокер: Writer не сдаётся после первой попытки, а ретраит запись с нарастающим интервалом, пока жив контекст; если брокер так и не поднялся, вызов завершается ошибкой контекста — context.DeadlineExceeded при таймауте или context.Canceled при отмене. Отсюда практическое правило: контекст на запись — не формальность, а ваш дедлайн; без него продюсер может крутиться в ретраях неограниченно долго. Вторая группа — ошибки брокера на корректной связи: запись в несуществующий топик без AllowAutoTopicCreation прилетает ошибкой Unknown Topic Or Partition, отказ в доступе к каналу — своим кодом. Все они приходят обёрнутыми, и вытащить код умеет errors.As с целью kafka.Error: значение этого типа само и есть код, его сравнивают с константами вроде kafka.UnknownTopicOrPartition. Третий источник — само сообщение: превышение лимита размера (BatchBytes в WriterConfig и потолок брокера) возвращает MessageTooLargeError с индексом проблемного сообщения — его выбрасывают из пачки и отправляют отдельно, а не глушат всю пачку целиком. Каждой группе — свой ретрай и свой алерт: это тема двенадцатого урока, здесь важно видеть карту. Хорошая новость: все три группы ловятся в одном месте — ошибкой, возвращённой из WriteMessages, и различаются уже по кодам, поэтому единый обработчик с логированием и метрикой покрывает весь спектр.

writer.Close() в продюсере — не церемония, а часть контракта. Close дожидается отправки накопленных батчей, завершает внутренние горутины и закрывает пул соединений: без него последние сообщения могут остаться в памяти процесса при завершении, а «молчаливая потеря» — худший из видов отказа для шины событий. Канонический порядок такой: Writer создаётся один на процесс, defer Close ставится рядом с конструктором, а при остановке сервиса закрывается явно после того, как обработчики закончили работу — приёмы выключения без потерь собраны в уроке про грабли продакшена. Обратная сторона долгожизни — конфигурация: Writer настраивается один раз и не пересобирается на ходу, поэтому всё, что может меняться, — топики в сообщениях, ключи и значения; сам Writer и его конфиг остаются неизменными. Для тестов Writer инжектится интерфейсом, как любая зависимость, — приём из урока про тесты Kafka. Правило на память: один Writer, длинная жизнь, аккуратный Close. И держите конфиг рядом с кодом запуска: полдня дебага уходит на то, чтобы найти, где именно продюсер настроен иначе, чем вы думали.