Reader API: чтение с оффсета

Конфигурация зеркальна групповой: в ReaderConfig вместо GroupID указывается Partition — конкретный номер партиции топика. Правило взаимной исключённости жёсткое: NewReader паникует, если заданы и Partition, и GroupID одновременно — документация формулирует «either Partition or GroupID may be specified, but not both», и паника на старте это лучший способ узнать о конфликте. Один reader в этом режиме читает ровно одну партицию: параллелизм получаете вручную, по reader на партицию, и распределение партиций по инстансам — тоже ваша работа, а не работа брокера. Взамен исчезают все групповые механики: никаких пульсов, генераций и перетасовок; процесс умер — чтение остановилось, поднялся — продолжили с той позиции, которую сами установили. Для разовой стартовой точки есть и поле конфига StartOffset — оно принимает те же константы FirstOffset и LastOffset, его дефолт FirstOffset, и оно экономит вызов SetOffset, когда позиция известна ещё до создания reader'а. Именно за предсказуемость groupless-режим любят в утилитах: реплей-джобы, аудиты, переливы между топиками. Цена — самостоятельность: никто не закоммитит позицию за вас, и если сервис упал, продолжение с места остановки нужно организовать явно — хранить оффсет там же, где и результат обработки.

SetOffset принимает три вида значений, и все три полезны в разных задачах. Константа kafka.FirstOffset — минус два — переводит позицию на самый старый доступный оффсет партиции; с включённым ретеншеном «старый» значит «самый ранний из живых», а не ноль. Константа kafka.LastOffset — минус один — ставит reader на хвост: следующий ReadMessage заблокируется и вернёт первое новое сообщение, а не последнее лежащее; чтобы перечитать лежащее, нужен конкретный номер. Третья форма — число: SetOffset(42) позиционирует точно на оффсет 42, и следующий ReadMessage вернёт именно его. Метод работает на лету: позицию можно двигать между чтениями, двигаясь по истории туда-обратно, и каждое перемещение видно в логе: reader печатает «setting the offset of the kafka reader … from … to …», так что мухлеж с позицией сам себя выдаёт в логах. Два пограничных случая из исходников: после Close SetOffset возвращает io.ErrClosedPipe, а позиция меньше первого живого оффсета при чтении аккуратно доезжает до него — reader печатает в лог, сколько сообщений он пропустил прыжком. Историческая ремарка для тех, кто встретит старые примеры: в версии 0.2.0 значения минус единицы и минус двойки менялись местами, поэтому константы по именам — не стиль, а защита от копипасты с призраками.

У groupless reader'а есть полный комплект измерений, который групповому недоступен. Offset() возвращает текущую позицию — абсолютный оффсет, на котором reader стоит. Lag() отдаёт кэшированный лаг: разницу между концом партиции и позицией, обновляемую по ReadLagInterval, который в конфиге можно отключить отрицательным значением. ReadLag(ctx) — свежий замер по сети: метод сходит к брокеру, узнаёт последний оффсет и считает лаг прямо сейчас; удобно для точечных проверок и метрик. Нюанс: при отрицательной позиции ReadLag возвращает ноль — «лаг неопределён, пока чтение не началось», и это стоит помнить, вызывая метод до первого чтения. Для группового режима всё это недоступно честно и по-разному: Offset и Lag возвращают минус единицу, а ReadLag — ошибку errNotAvailableWithGroup; лаг группы считается по агрегированным данным брокера, о чём говорит двадцать первый урок. Из полезных полей сообщения в этом режиме отметим HighWaterMark: каждое прочитанное сообщение несёт конец партиции на момент чтения — дешёвый способ оценить остаток без отдельного запроса. Все эти измерения превращают groupless reader в прибор: скрипт мониторинга, отладочный хвост, счётчик непрочитанного — всё это десять строк кода.

Соберём список задач, где groupless reader — правильный инструмент, а не костыль. Реплей и backfill: перечитать диапазон оффсетов или день по времени после миграции или смены правил; группа для этого не годится — её позиция нельзя трогать без последствий. Аудит и отладка: пройти партицию с начала с чужой логикой, посмотреть живые события через хвост, сверить, что реально лежит в топике. Карантин: перечитать события из карантинного топика девятнадцатого урока с новыми правилами обработки — позиция карантинного джоба полностью под его контролем. Ручная раздача: когда по reader на партицию и своё распределение дают контроль, которого группа не даёт, — например, привязка партиций к мощностям инстансов. Мониторинг: ReadLag по каждой партиции в метрики — прибор из двадцать первого урока. Общая деталь всех сценариев — чекпоинт: утилита реплея обычно печатает последний обработанный оффсет в файл или переменную процесса запуска, чтобы после сбоя продолжить с него через SetOffset с числом, а не с начала; это тот же контракт, что и коммиты группы, только хранилище позиции — ваше. И обратный список: штатный поток событий с масштабированием — работа групп из седьмого урока; писать свой раздальщик партиций поверх groupless readers, чтобы избежать группы, — вот это уже костыль, который придётся поддерживать самому.

Иногда нужен уровень ниже: kafka.DialLeader открывает прямое соединение с лидером конкретной партиции — DialLeader(ctx, "tcp", "localhost:9092", topic, partition) возвращает сырой *Conn. У него есть свои Read, ReadMessage, ReadBatch и своя навигация по оффсетам: Seek с Offset, а также ReadFirstOffset, ReadLastOffset и ReadOffsets — по сути, это транспорт без планировщика, пульсов и прочей обвязки Reader. Годится для точечных инструментов: быстро прочитать несколько сообщений, снять оффсеты, сделать разовую миграцию. Для сервисного кода Reader остаётся правильным выбором: он переживает переезды лидера, переконнекты и держит батчинг чтения — сырой Conn ничего этого не делает за вас. Практический ориентир: если утилита одноразовая и живёт секунды, DialLeader быстрее собрать и проще выбросить; если инструмент живёт в проде дольше минуты — берите Reader и получайте отказоустойчивость бесплатно.