Первый консьюмер: Reader

Консьюмер начинается с kafka.NewReader, который принимает ReaderConfig и паникует, если конфиг невалиден. Минимальный набор — Brokers, Topic и либо Partition, либо GroupID: это два режима работы, взаимоисключающие на уровне Validate. Чтение конкретной партиции — режим без состояния: вы говорите, откуда читать, и Reader не коммитит ничего. Чтение группой — режим с координацией, о нём следующий урок. Дефолты конфига стоят того, чтобы их знать: MinBytes равен 1 — брокер отвечает, как только есть хоть байт данных; MaxBytes — примерно мегабайт на батч, чтобы ответ брокера не разорвал память; MaxWait — 10 секунд ожидания данных в незаполненном батче. QueueCapacity внутренней очереди — 100 сообщений: они выравнивают скорость fetch и скорость вашего обработчика. Рядом стоят и менее заметные поля: MaxAttempts ограничивает тремя число попыток соединения, прежде чем ошибка уйдёт наружу, а ReadBatchTimeout — десятью секундами выборку из уже полученного батча. Эти цифры объясняют поведение, которое без них выглядит загадкой: почему консьюмер на пустом топике просыпается раз в десять секунд и почему большой MinBytes на низконагруженном топике добавляет задержку — брокер копит данные до минимума, прежде чем ответить. Для первого kafka reader оставьте дефолты как есть: они разумны, а тюнинг батчей — дело измерений, о которых поговорим в десятом уроке на продюсерской стороне.

ReadMessage читает и возвращает следующее сообщение, блокируя горутину, пока его нет: на живом потоке это микросекунды, на пустой партиции — хоть до утра. Отменить ожидание можно контекстом — так останавливают читателя при выключении сервиса. Возвращаемое значение — kafka.Message, и его поля — это адрес события в логе: Topic и Partition говорят, откуда, Offset — позиция, Key и Value — байты, Time — время записи брокером, Headers — служебные пары ключ-значение, которые продюсер приложил сам: трейс-идентификаторы и метки маршрутизации живут именно там, а не в Value. Время записи брокером, а не обработки: на нём строится задержка сквозной доставки. Поле HighWaterMark показывает последний оффсет, видимый брокеру, — из него считается лаг. Ошибку ReadMessage оборачивает в префикс fetching message, поэтому проверяйте через errors.Is и errors.As, а не сравнением строк. По устройству метод — тонкая обёртка: внутри это FetchMessage, который при группе сопровождается CommitMessages, и ошибка коммита приедет с префиксом committing message. Особый случай — io.EOF: его Reader возвращает после Close, и это штатный сигнал остановки, а не сбой. Если программа прочитала лог до конца, ReadMessage не вернёт управление с ошибкой — он повиснет и будет ждать новых данных, потому что чтение журнала в Kafka бесконечно по определению. Отсюда практическое правило вычитки истории: ограничивайте её не чтением до конца, а оффсетом или контекстом с дедлайном из восемнадцатого урока — иначе процесс застрянет в бесконечном ожидании после последнего события.

Цикл консьюмера живёт весь срок жизни сервиса, и единственная ветка выхода в нём — ошибка от ReadMessage. Обработка этой ошибки — место, где новички ломают продакшен. Первый грех: логировать ошибку и продолжать, не отличая io.EOF от сбоя. После Close такой цикл превращается в горловину: Reader мёртв, ReadMessage мгновенно возвращает io.EOF, лог раздувается на гигабайты, процесс жжёт CPU впустую. Второй грех — выход по любой ошибке без стратегии: сеть мигнула, брокер перезапустился, и читатель умер вместе с сервисом, хотя kafka-go умеет переподключаться сам. Разумная схема выглядит так: io.EOF — тихий выход, контекст отменён — тихий выход, остальное — лог и либо выход с падением процесса, чтобы оркестратор перезапустил, либо пересоздание Reader с паузой. Контекст тут — главный инструмент: отменённый контекст гасит блокирующий вызов мгновенно, поэтому выключение сервиса выглядит как отмена, а не как убийство горутин с риском недоделанной обработки. Внутри Reader между попытками fetch уже стоит бэкофф от 100 миллисекунд до секунды, так что при живом Reader шторма запросов не будет. Определять политику перезапуска заранее — часть контракта консьюмера: ошибка в этой ветке видна только в час ночи, и смена решения на ходу стоит дороже десяти минут проектирования.

Без группы вы владеете позицией напрямую, и это стоит потренировать на живом топике. SetOffset принимает kafka.FirstOffset, kafka.LastOffset или конкретное число — это переигрывание истории из урока про устройство брокера в чистом виде. Offset возвращает текущую позицию — следующий оффсет, который будет прочитан. Lag — расстояние до конца лога, вычисленное по последнему fetch: на живом потоке цифра достаточно свежая, на пустом устаревает, потому что fetch не выполняется. Для честной цифры есть ReadLag — отдельный запрос к брокеру, который вернёт актуальное отставание; ценой будет лишний round-trip, поэтому зовут его по таймеру, а не на каждое сообщение. С GroupID картина другая: SetOffset и ReadLag вернут ошибку unavailable when GroupID is set, а Offset и Lag — минус единицу, потому что позиция группы живёт на брокере, и клиенту её не отдают. Наконец, Stats отдаёт структуру ReaderStats со счётчиками Fetches, Messages, Bytes, Errors и Rebalances — их стоит экспортировать в мониторинг, прежде чем городить свой учёт: встроенные метрики уже считают то же самое, а параллельный учёт в своих обёртках — только источник расхождений. Разница между ReadLag по таймеру и встроенным Lag видна на графике за минуты — на тихом топике встроенная цифра стоит на месте и обманывает, пока не придёт новое сообщение.

Честно о формате: песочница сайта сеть не поднимает, поэтому этот код запускается локально: Go и Docker с брокером из третьего урока, два терминала — в одном продюсер из четвёртого, в другом этот Reader. Вы увидите своими глазами, как Offset растёт, а Lag тает, и что ReadMessage на пустом топике молчит, пока вы не пришлёте сообщение. Интерактивы ниже — тематические программы на чистом Go: они тренируют позицию, ошибки и лаг на эмуляции лога, и их вывод совпадает с тем, что покажет настоящий брокер на тех же данных. Если брокер ещё не поднят — интерактивов хватит для проверки в конце урока, а живой прогон останется пятиминуткой вместо вечера непонимания.