Прямой ответ
Отставание консумента — разница между последним смещением в партиции и последним обработанным смещением группы. Метрика показывает, сколько сообщений накопилось без обработки.
lag = log_end_offset − committed_offset
Получение значений через клиент:
from kafka import KafkaConsumer, TopicPartition
c = KafkaConsumer(group_id="orders-processor", bootstrap_servers=BOOTSTRAP)
parts = c.partitions_for_topic("orders")
tps = [TopicPartition("orders", p) for p in parts]
c.assign(tps)
end = c.end_offsets(tps)
committed = {tp: c.committed(tp) or 0 for tp in tps}
for tp in tps:
print(tp.partition, "lag =", end[tp] - committed[tp])
Отставание считается по каждой партиции отдельно, поскольку партиции обрабатываются независимо; о партициях — в статье про распределение сообщений.
| Состояние | Значение lag | Интерпретация |
|---|---|---|
| Норма | Близко к нулю, колеблется | Обработка успевает за записью |
| Пик нагрузки | Растёт, затем снижается | Временное накопление |
| Устойчивый рост | Монотонно увеличивается | Обработка медленнее записи |
| Перекос | Одна партиция выше остальных | Перекос ключа либо сбой консумента |
Отставание измеряется в сообщениях, поэтому интерпретация зависит от интенсивности записи. Одно и то же значение означает минуты задержки при низкой записи и часы при высокой.
Почему отставание в сообщениях недостаточно для оповещения
Порог оповещения, заданный числом сообщений, даёт ложные срабатывания при росте трафика и пропуск проблемы при низком трафике.
Рост трафика без сбоя. Интенсивность записи увеличилась, и отставание выросло в сообщениях, но время обработки одного сообщения не изменилось. Оповещение срабатывает, хотя обработки успевает.
Снижение трафика при сбое. Запись почти прекратилась, поэтому отставание в сообщениях растёт медленно, хотя обработка остановлена. Оповещение не срабатывает в течение длительного времени.
Практическое следствие: вторая метрика — время, соответствующее отставанию. Она вычисляется по отметке времени сообщения, на которое указывает обработанное смещение.
задержка = now − timestamp(сообщения по committed_offset)
Задержка в секундах показывает, насколько старые данные обрабатываются, и не зависит от интенсивности записи. О метриках отставания очередей — в статье про мониторинг очередей.
Третья полезная метрика — скорость обработки в сообщениях в секунду в сравнении со скоростью записи. Превышение скорости записи над скоростью обработки означает, что отставание будет расти независимо от текущего значения.
Четыре причины устойчивого роста отставания.
Обработка медленнее записи. Время обработки одного сообщения либо число консументов недостаточны.
Перекос по партициям. Одна партиция получает основную долю сообщений, и назначенный на неё консумент не справляется.
Перебалансировка группы. Частые перебалансировки останавливают обработку, поэтому накопление происходит во время перераспределения; о перебалансировке — в статье про consumer groups.
Сбой внешнего сервиса. Обработка ожидает ответа внешнего сервиса, и задержка ответа снижает скорость обработки.
Связь отставания со сроком хранения данных
Отставание ограничено сроком хранения журнала: сообщения, не обработанные до удаления, утрачиваются безвозвратно. Поэтому отставание — не только метрика задержки, но и индикатор риска потери.
Критическое значение отставания определяется не требованием к своевременности, а оставшимся сроком хранения. Если интенсивность записи составляет тысячу сообщений в минуту, а срок хранения — сутки, то критическое отставание соответствует примерно полутора миллионам сообщений. Превышение означает, что часть необработанных сообщений будет удалена.
Практическое следствие: второй порог оповещения настраивается на долю срока хранения, а не на абсолютное число сообщений. Оповещение срабатывает до того, как потеря становится возможной, и оставляет время на масштабирование обработки.
Отдельный аспект — поведение при длительной остановке консумента. После возобновления обработки часть сообщений может оказаться уже удалённой, поэтому восстановление требует чтения с доступного начала журнала и фиксации факта пропуска в журнале обработки.
Практика: настройка мониторинга отставания
1. Собирайте отставание по каждой партиции. Суммарное значение скрывает перекос, при котором одна партиция отстаёт при нулевом отставании остальных.
2. Добавьте метрику задержки в секундах. Значение в сообщениях не отражает фактическую задержку обработки при изменении трафика.
3. Сравнивайте скорость обработки со скоростью записи. Превышение скорости записи означает неизбежный рост отставания.
4. Установите порог оповещения по задержке, а не по числу сообщений. Порог в секундах соответствует требованиям к своевременности данных.
5. Оповещайте о перебалансировках. Частые перебалансировки указывают на нестабильность группы и приводят к накоплению.
6. Проверяйте распределение отставания по партициям. Перекос указывает на проблему с ключом либо с отдельным консументом.
7. Фиксируйте базовое значение отставания. Отклонение от базового уровня выявляет проблему раньше абсолютного порога.
Рекомендации:
- Различайте временный пик и устойчивый рост. Пик после всплеска трафика снижается самостоятельно, поэтому оповещение настраивается на длительность превышения.
- Учитывайте срок хранения данных. Отставание, превышающее срок хранения, означает потерю необработанных сообщений.
- Не масштабируйте число консументов сверх числа партиций. Лишние консументы не участвуют в обработке, поэтому масштабирование требует увеличения партиций.
- Проверяйте скорость обработки при изменении кода. Замедление обработки одного сообщения накапливает отставание постепенно; о настройке оповещений — в статье про Alertmanager.
- Документируйте ожидаемое отставание. Значение фиксируется как базовый уровень, отклонение от которого требует разбора.
Практический сценарий: отставание росло без оповещения
Контекст: данные в отчёте отставали на несколько часов, при этом оповещения о накоплении не срабатывали.
- Диагностика. Отставание в сообщениях составляло несколько тысяч, что находилось ниже установленного порога. Интенсивность записи была низкой, поэтому накопление шло медленно. Задержка по отметке времени составляла несколько часов.
- Причина. Порог оповещения задан числом сообщений. При низкой интенсивности записи несколько тысяч сообщений соответствуют часам задержки, но не превышают порог, поэтому проблема не обнаруживалась.
- Действия. (а) Добавили метрику задержки в секундах, вычисляемую по отметке времени сообщения по обработанному смещению; (б) перенесли порог оповещения на задержку, соответствующую требованию к своевременности отчёта; (в) добавили сравнение скорости обработки со скоростью записи как ранний индикатор; (г) нашли причину замедления: обработка выполняла запрос к внешнему сервису без таймаута, и замедление ответа снижало скорость; (д) установили таймаут и повтор с отсрочкой, вернув скорость обработки к базовому уровню.
- Результат. Отставание обнаруживается по задержке независимо от интенсивности записи, а причина замедления устранена.
Вывод: порог оповещения по числу сообщений не отражает фактическую задержку при изменении интенсивности записи. Метрика задержки в секундах и сравнение скоростей обработки и записи выявляют накопление независимо от объёма трафика.
FAQ
Как считается отставание консумента? Как разница между последним смещением в партиции и последним обработанным смещением группы, отдельно по каждой партиции.
Почему оповещение по числу сообщений ненадёжно? Одно и то же число сообщений соответствует разной задержке при разной интенсивности записи.
Что означает отставание в одной партиции при нулевом в остальных? Перекос распределения по ключу либо сбой консумента, назначенного на эту партицию.
Может ли отставание привести к потере данных? Да, если оно превышает срок хранения данных в топике: необработанные сообщения удаляются.
Источники: Apache Kafka monitoring, Prometheus client libraries, PromQL basics — ссылки в шапке.