--- title: "Мониторинг: лаг консьюмера" description: "Лаг консьюмера в kafka-go: ReadLag, статистика Writer, алерты на отставание. Как считать лаг группы и партиции и следить за ним фоновой горутиной." course: "Kafka в Go: продюсеры и консьюмеры на kafka-go" url: "https://golangmaster.ru/courses/kafka-go/lessons/21-monitoring-lag/" published: "2026-09-19" updated: "2026-09-29" minutes: 18 --- # Мониторинг: лаг консьюмера *Курс «Kafka в Go: продюсеры и консьюмеры на kafka-go», урок 21 из 25. ~18 минут. Обновлено 29.09.2026. Источник: https://golangmaster.ru/courses/kafka-go/lessons/21-monitoring-lag/* Лаг консьюмера — главная метрика любого kafka-потока: она отвечает не на вопрос «работает ли код», а на вопрос «успевает ли система за данными», и в kafka-go за неё отвечают ReadLag и Lag у Reader плюс Stats у Writer. Разберём, что именно означает число лага, почему ноль тоже сигнал, как получить лаг для одиночного ридера и почему для группового консьюмера API честно откажется работать, что показывать статистика Writer и как собрать фоновую горутину с тикером, которая ведёт лог и питает алерты на отставание. Продюсер исправно пишет, консьюмер жив и даже коммитит офсеты — а вечером выясняется, что заказы обрабатывались с задержкой в час. Код не падал, ошибок не было, потребитель просто работал медленнее потока событий. Единственная метрика, которая ловит такую ситуацию заранее, — лаг консьюмера: разница между последним офсетом партиции и офсетом, на котором стоит группа или ридер. Это число означает «столько сообщений ещё не разобрано», и оно говорит о здоровье потока больше, чем любой лог. Седьмой урок строил группы, восьмой — коммиты офсетов, тринадцатый разбирал ребалансировку; все они манипулировали механизмом, но не отвечали на вопрос владельца сервиса «как быстро мы догоняем конец очереди прямо сейчас». Здесь отвечаем: сначала арифметика лага, потом инструменты kafka-go для его измерения, затем паттерн фонового мониторинга и правила алертов, при которых сообщение приходит до того, как позвонил клиент. ### Лаг консьюмера: разница двух офсетов Формула лага двухсложная. У каждой партиции есть лог с последним офсетом — high watermark, число записей, которое брокер уже принял. У консьюмера есть собственная позиция — офсет последнего разобранного сообщения, который он же и коммитит. Лаг партиции равен разности: последний офсет минус текущий офсет группы. Ридер без GroupID из четырнадцатого урока хранит эту позицию сам, групповой консьюмер из седьмого — в координаторе, но смысл числа одинаков: столько событий легло в топик, а в хендлер ещё не попало. Из формулы следуют обе стороны жизни лага. Растущий лаг означает, что в единицу времени в партицию пишется больше, чем консьюмер успевает разобрать: не хватает партиций, упала скорость хендлера, задерживает внешняя система. Отрицательного лага не бывает, а вот лаг, застрявший на нуле, — не всегда победа: он говорит лишь, что очередь пуста в момент замера, и ничего — о том, жив ли сам потребитель и коммитит ли он позиции. Именно поэтому лаг смотрят в динамике, а не одним срезом, и всегда вместе с признаками жизни процесса. ### Kafka lag мониторинг: что считать нормой В сыром виде лаг — штуки, и порог «лаг больше тысячи» ничего не значит: для топика с десятью событиями в сутки это осень, для платежного потока — полминуты. Kafka lag мониторинг начинается с перевода лага во время: лаг, умноженный на среднюю скорость обработки одной партиции, даёт минуты задержки, и алерт ставится именно на задержку — «фоновая обработка отстаёт больше чем на пять минут». Второй обязательный график — скорость изменения лага. После деплоя лаг, выросший ступенькой и вернувшийся к нулю, означает паузу на рестарт и ребаланс из тринадцатого урока; лаг, выросший и не возвращающийся, означает, что новый код медленнее старого. Ноль, продержавшийся часы при живом потоке продюсера, — тоже не повод для радости, а сигнал смотреть коммиты: либо консьюмер перечитывает то же место с кривым стартовым офсетом, либо коммиты задерживаются, и позиция группы врёт. Отдельно честный вывод: единственного числа «лаг» мало — мониторят лаг, скорость его роста и факт коммитов, и только связка этих линий отличает живую систему от тихо заглохшей. ### ReadLag: лаг одиночного ридера ```go reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{"localhost:9092"}, Topic: "payments", Partition: 0, // GroupID не задан: методы лага доступны }) defer reader.Close() lag, err := reader.ReadLag(ctx) if err != nil { log.Printf("lag: %v", err) } log.Printf("лаг партиции: %d, кэш: %d", lag, reader.Lag()) ``` ReadLag синхронно спрашивает у брокера последний офсет партиции и вычитает из него офсет последнего сообщения, отданного ReadMessage, — результат свежий, но стоит сетевого запроса. Lag() читает внутренний кэш, который Reader обновляет в фоне, поэтому бесплатен, но отстаёт на интервал обновления. Пока текущий офсет ридера отрицательный, ReadLag возвращает ноль. ### Проверка: ReadLag у группового консьюмера **Вопрос:** Reader создан с GroupID: "payments-group". Что произойдёт при вызове ReadLag? - ✅ Вернётся ошибка errNotAvailableWithGroup: лаг недоступен при заданном GroupID - ReadLag вернёт суммарный лаг всех партиций группы - ReadLag вернёт лаг партиции, назначенной этому ридеру - Вызов заблокируется до ближайшего ребаланса В reader.go первой строкой ReadLag проверяет, работает ли Reader как групповой консьюмер, и возвращает errNotAvailableWithGroup — «unavailable when GroupID is set». Позиция группового консьюмера живёт в координаторе, а не в Reader, поэтому метод честно отказывается давать число, которого он не знает. Lag() в этой ситуации возвращает минус единицу. ### errNotAvailableWithGroup: лаг и группы Ограничение из квиза — не досадная недоделка, а следствие архитектуры. Ридер без GroupID знает свою позицию сам: он хранит текущий офсет в памяти и потому способен посчитать лаг в один запрос. Групповой консьюмер из седьмого урока позицию не выбирает — её назначает координатор, коммиты лежат в брокере, и после ребалансировки партиции переехали бы к другим членам в любой момент. Число в памяти отдельного Reader ничего бы не значило, поэтому kafka-go возвращает errNotAvailableWithGroup, а Lag() — минус единицу. Что делать, если поток живёт группой, а лаг видеть нужно? Три пути. Первый — снять статистику топика напрямую через Client: разность последнего офсета партиции и закоммиченного офсета группы, метод, который мы разберём в восемнадцатом уроке про контексты и таймауты. Второй — вывести лаг на сторону брокера: kafka-lag-exporter и консольные утилиты кластера считают его из коммитов. Третий — держать для наблюдения отдельный ридер без GroupID, читающий тот же топик с LastOffset: он не мешает группе, потому что не состоит в ней. Практическое правило: ReadLag — для ручных и тестовых потоков, групповой лаг — за пределами kafka-go, в метриках кластера. > **Живое место: лаг по одной партиции — не лаг потока.** ReadLag у ридера с фиксированной партицией отвечает ровно за неё, и смешивать его с лагом всего топика — классическая ошибка графиков. Топик из двенадцати партиций с лагом ноль на партиции ноль может тонуть в партиции семь, где после кривого ключа из пятого урока скопились заказы одного крупного клиента. Вторая ловушка: ReadLagInterval у Reader по умолчанию равен минуте, и кэшный Lag() между обновлениями показывает старое число — в алертах это выглядит как запаздывающая реакция. Инцидент-паттерн: дежурный видит ноль, успокаивается, а партиция семёрка в это время растёт. Лаг собирается по всем партициям потока, а не по одной удобной. ### Writer.Stats: статистика стороны записи ```go st := writer.Stats() log.Printf("topic %s: writes=%d msgs=%d errors=%d retries=%d", st.Topic, st.Writes, st.Messages, st.Errors, st.Retries) log.Printf("batch avg=%d write avg=%v", st.BatchSize.Avg, st.WriteTime.Avg) ``` Stats() снимает срез WriterStats: Writes — записи, Messages — сообщения, Bytes — объём, Errors — счётчик ошибок, Retries — ретраи, BatchSize и BatchBytes — сводка размеров батчей, WriteTime — распределение времени записи с полями Avg, Min, Max. Счётчики агрегируются внутри Writer, поэтому метрику снимают периодически, а не по событию. ### Проверка: что растёт в WriterStats **Вопрос:** После деплоя Writer.Stats показывает: Errors растёт, Messages стоит на месте. Какой вывод корректен? - ✅ Записи отклоняются: брокер отвечает ошибками, до записи сообщений дело не доходит - Writer разучился считать сообщения: это баг kafka-go - Лаг консьюмера вырос: статистика Writer связана с чтением - BatchSize достиг MaxBatchSize и Writer встал Messages — счётчик фактически отправленных сообщений, Errors — неудачных попыток. Рост Errors при стоящих Messages означает, что пачки уходят и возвращаются с ошибкой: топик отсутствует, кластер недоступен, сработали NotEnoughReplicas из пятнадцатого урока. Лаг консьюмера здесь ни при чём: WriterStats описывает только сторону записи. ### Что вернёт ReadLag у свежего ридера? ```go // Reader без GroupID, только что создан: // ReadMessage ни разу не вызывался, // StartOffset по умолчанию — дефолтный. lag, err := reader.ReadLag(ctx) log.Printf("lag=%d err=%v", lag, err) ``` **Что выведет программа?** - ✅ lag=0: текущий офсет ридера отрицательный, считать разность не с чего - lag=-1: ридер ещё ничего не читал - errNotAvailableWithGroup: офсет не назначен - Лаг всего топика: ридер обязан спросить брокера Документация ReadLag фиксирует: функция возвращает ноль, когда текущий офсет ридера отрицательный. Пока ReadMessage не отдал ни одного сообщения, позиция не определена, и разность «конец партиции минус позиция» не имеет смысла. Минус единицу в этой ситуации возвращает не ReadLag, а Lag() у группового ридера — чужой признак, легко перепутать. ### Фоновый мониторинг: горутина с тикером Лаг, который смотрят руками, мониторингом не является: его забудут посмотреть именно в тот вечер, когда он вырастет. Паттерн фонового мониторинга — отдельная горутина с тикером, живущая в контексте приложения. Она периодически снимает лаг всех наблюдаемых ридеров и партиций, пишет его в лог или отправляет в систему метрик, и умирает вместе с контекстом, не оставляя утечек. Три детали делают горутину рабочей, а не декоративной. Во-первых, свой таймаут на каждый замер: ReadLag ходит в брокер, и замер не должен зависеть от чужих таймаутов — приёмы из восемнадцатого урока. Во-вторых, ошибка замера — тоже метрика: недоступный брокер для мониторинга выглядит как всплеск ошибок, а не как ноль лага, иначе дежурный увидит обманчивое благополучие. В-третьих, замер должен обходить все партиции потока, а не одну. Занятно, что внутри Reader уже живёт похожая механика: ReadLagInterval запускает фоновую горутину, которая с тикером обновляет кэш для Lag(), — наш паттерн повторяет её, но во внешний мир. Для Async-Writer из десятого урока та же горутина снимает Stats: без неё ошибки записи невидимы в принципе. ### Правила мониторинга лага - Лаг измеряется в сообщениях, алерты ставятся на минуты задержки: лаг умножить на среднюю скорость обработки. - Мониторят три линии: абсолютный лаг, скорость его роста и факт коммитов — по одной линии диагноз не ставится. - ReadLag работает только у ридера без GroupID; для групп лаг снимают на стороне брокера или отдельным ридером. - Замер лага получает свой таймаут: недоступный брокер — это метрика ошибок, а не ноль лага. - Writer.Stats снимают по тикеру для Async-Writer: иначе ошибки записи не видны никому. - Рост лага после деплоя — первичный сигнал, что новый код медленнее предыдущего. ### Алерты: лаг в минутах и рост после деплоя Арифметика алерта: средняя скорость обработки партиции — десять событий в секунду, лаг пять тысяч — восемь минут задержки. Порог настраивается на время, а не на штуки: «задержка больше пяти минут в течение трёх проверок подряд» — правило, которое не сработает от случайного всплеска и не промолчит при реальном отставании. Второе правило — ступенька после деплоя: короткий рост лага на рестарте с последующим возвратом к нулю — норма ребаланса; рост, который не возвращается за таймаут выката, открывает инцидент ещё до жалоб пользователей. Третье — зеркальный алерт на подозрительный ноль при живом продюсере: он ловит и застрявший консьюмер, и задержавшиеся коммиты, потому что и то и другое выглядит как позиция, не двигающаяся с места. Свяжем с ошибками чтения: если лаг растёт вместе с ретраями и ошибками из девятнадцатого урока, потребитель не медленный, а падающий — яд-сообщение, от которого группа циклится, живёт в списке пунктов о ядовитых сообщениях. Мониторинг лага без мониторинга ошибок чтения половину таких случаев примет за «просто не успеваем». ### Инструменты мониторинга одним столом | Инструмент | Что показывает | Нюанс | | --- | --- | --- | | Reader.Lag() | Кэш лага после последнего ReadMessage | Минус один при GroupID; отстаёт на ReadLagInterval | | Reader.ReadLag(ctx) | Свежая разность конца партиции и позиции ридера | Ошибка errNotAvailableWithGroup при GroupID; ноль при отрицательном офсете | | ReadLagInterval | Частота фонового обновления кэша лага | Дефолт — минута; задаётся в ReaderConfig | | Writer.Stats() | Writes, Messages, Errors, Retries, размеры батчей, время записи | Для Async-Writer — единственное окно в судьбу сообщений | | Метрики брокера | Лаг групповых консьюмеров по коммитам | Вне kafka-go: exporter или утилиты кластера | ### Соберите замер лага Допишите частоту фонового обновления лага в конфиге Reader и имя метода, который даёт свежую разность офсетов. ```go reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{"localhost:9092"}, Topic: "payments", Partition: 0, ReadLagInterval: 30 * time.___, }) lag, err := reader.___(ctx) if err != nil { log.Printf("lag: %v", err) } ``` Подсказка: ReadLagInterval — обычный time.Duration, а свежий лаг снимается методом ReadLag с контекстом. Правильные ответы в пропусках: `Second`, `ReadLag` ### Соберите горутину мониторинга лага Расставьте строки фоновой горутины: тикер, остановка, цикл с select, замер лага, обработка ошибки, лог числа и запуск рядом с ридером. Правильный порядок строк: 1. `ticker := time.NewTicker(30 * time.Second)` 2. `defer ticker.Stop()` 3. `for { // до отмены контекста приложения` 4. `select {` 5. `case <-ctx.Done(): return` 6. `case <-ticker.C:` 7. `lag, err := reader.ReadLag(ctx)` 8. `if err != nil { log.Printf("lag: %v", err); continue }` 9. `log.Printf("лаг консьюмера: %d", lag)` 10. `} // select` 11. `} // for` 12. `go monitorLag(ctx, reader) // запуск рядом с Reader` > **Про песочницу: где это запускается.** Все замеры из урока работают локально: брокер в Docker из третьего урока, пара Writer и Reader, горутина с тикером — и в терминале ползёт живой лог лага. Схема честности привычная: интерактивы разыгрывают арифметику офсетов на программах в чистом Go без сети, а продюсер, консьюмер и ReadLag против настоящего брокера запускаются только локально. Продакшен-складывание лага по всем партициям и exporter для групп — тема следующего этапа: сначала у себя в песочнице увидьте, как лаг растёт при остановленном консьюмере и как ноль наступает после его догоняющего рывка. ### Карточки: лаг и статистика - **Лаг консьюмера** — Разность последнего офсета партиции и текущего офсета группы или ридера: столько сообщений ещё не разобрано. - **Reader.ReadLag(ctx)** — Свежий лаг запросом к брокеру. При заданном GroupID — errNotAvailableWithGroup; при отрицательном текущем офсете возвращает ноль. - **Reader.Lag()** — Кэш лага последнего прочитанного сообщения, обновляемый в фоне. Для группового ридера возвращает минус единицу. - **ReadLagInterval** — Частота фонового обновления лага в ReaderConfig; при нуле NewReader подставляет одну минуту. - **Writer.Stats()** — Срез WriterStats: Writes, Messages, Bytes, Errors, Retries, BatchSize, WriteTime. Для Async-Writer — единственный доступ к судьбе записей. - **Алерт на лаг** — Ставится на минуты задержки: лаг умножить на среднюю скорость обработки, плюс правило ступеньки после деплоя и алерт на подозрительный ноль. Итог части: лаг — не прихоть observability-команд, а единственное число, показывающее, успевает ли система за своими данными. В kafka-go ридер без группы даёт его двумя методами — свежим ReadLag и кэшным Lag(), групповой консьюмер отвечает errNotAvailableWithGroup и отправляет за лагом к метрикам брокера, а Writer.Stats закрывает сторону записи теми же счётчиками и распределениями. Дальше по курсу: [урок про группы](/courses/kafka-go/lessons/07-gruppy-consjumerov/) объяснил, почему позиция группы не принадлежит Reader; [урок про коммиты](/courses/kafka-go/lessons/08-kommit-ofsetov/) — как закоммиченный офсет становится второй половиной формулы лага; [урок про ребаланс](/courses/kafka-go/lessons/13-rebalansirovka/) — откуда на графике берутся ступеньки после деплоя; а [урок про таймауты](/courses/kafka-go/lessons/18-kontekst-tajmauty/) — как снимать метрики так, чтобы замер не зависал вместе с замеряемым. Начните в песочнице: поднимите локально брокер, запустите Writer, погасите Reader на минуту — и посмотрите, как число лага честно расскажет о вашей паузе. ## Коротко: что запомнить - Лаг консьюмера — разность последнего офсета партиции и офсета группы: столько сообщений ещё не разобрано. - Главный порог алерта — минуты задержки: лаг умножить на среднюю скорость обработки партиции. - Ноль лага при живом продюсере — тоже сигнал: застрявший ридер или задержавшиеся коммиты. - ReadLag даёт свежий лаг и отказывается работать у группы: errNotAvailableWithGroup при заданном GroupID. - Lag() читает кэш, обновляемый фоном раз в ReadLagInterval, и возвращает минус единицу для группы. - Writer.Stats отдаёт WriterStats: Writes, Messages, Errors, Retries, BatchSize, WriteTime — окно в сторону записи. - Фоновый мониторинг — горутина с тикером и собственным таймаутом замера по всем партициям. - Рост лага после деплоя и лаг с ошибками чтения — разные диагнозы: медленный код против яд-сообщения. ## Домашнее задание Поднимите локальный стенд из третьего урока и соберите пару Writer и Reader на топик payments с одной партицией, Reader — без GroupID. Напишите горутину monitorLag по образцу урока с тикером в тридцать секунд и контекстом приложения. Остановите Reader на две минуты, затем запустите снова: убедитесь, что лог лага вырос во время паузы и вернулся к нулю после рывка, и объясните по формуле «лаг на скорость», сколько это секунд задержки. Затем добавьте второй Writer с Async: true и снимайте Stats в том же тикере: добейтесь, чтобы Errors в WriterStats оказались на графике рядом с лагом. Отдельно создайте Reader с GroupID и зафиксируйте ответ ReadLag — ошибку errNotAvailableWithGroup, — записав в комментариях, где в таком случае живёт правда о лаге группы. ## Вопросы и ответы по уроку ### Почему ReadLag не работает с GroupID и это не починят? Потому что позиция группового консьюмера не принадлежит Reader: её назначает координатор, коммит хранится у брокера, а после ребаланса партиция вообще переедет к другому члену группы. Любое число в памяти конкретного Reader было бы вымыслом, поэтому library возвращает errNotAvailableWithGroup, а Lag() — минус единицу. Лаг группы считается из закоммиченных офсетов и текущих концов партиций — то есть на стороне брокера: exporter или утилиты кластера, либо самописный замер через Client, который мы разберём в восемнадцатом уроке. ### Lag() показывает ноль, но сообщения точно есть. Так как оно вообще работает? Lag() — это кэш, а не запрос: он возвращает разность, которую Reader посчитал в последний раз, когда отдал вам сообщение или когда фоновая горутина обновила её по ReadLagInterval. Пока ReadMessage не вызывался или брокер недоступен, кэш ничего не знает. Для решения, а не для индикации, используйте ReadLag — он ходит к брокеру за свежими офсетами в момент вызова. И проверьте партицию: у ридера с фиксированной партицией кэш отвечает только за неё. ### Какой порог лага ставить в алерте? Не в штуках, а во времени: посчитайте среднюю скорость обработки одной партиции и выразите лаг в минутах задержки. Типовой старт — алерт на пять минут задержки при двух-трёх проверках подряд, плюс отдельное правило на ступеньку после деплоя и отдельное на подозрительный ноль при живом продюсере. Порог в штуках без привязки к скорости либо спит на мелком топике, либо воет на большом; в минутах он одинаковый для обоих.