Устранение распространенных проблем с отставанием потребителя Kafka с помощью консольных команд

Отставание потребителя — это первое число, которое проверяют большинство операторов Kafka, когда конвейер работает медленно, но его также легче всего неправильно интерпретировать. Группа может показывать отставание в миллион записей, потому что один нижестоящий API зависает по таймауту, потому что развертывание оставило половину потребителей офлайн, потому что одна партиция горячее остальных или потому что приложение здорово и просто догоняет после запланированной паузы. Команды просты. Суждения вокруг них — вот где выигрываются или проигрываются инциденты.

Это руководство фокусируется на пути командной строки, который я использую во время инцидента с отставанием: описать группу, сравнить партиции, подтвердить, живы ли потребители, решить, растет отставание или уменьшается, и только затем рассмотреть сброс смещения. Сбросы смещений включены, потому что они иногда необходимы, но они не являются лекарством от медленного потребителя. Они либо пропускают работу, либо воспроизводят ее. Относитесь к ним как к операционному решению, а не как к исправлению производительности.

Понимание отставания потребителя Kafka

В Kafka сообщения организованы в топики, которые дополнительно делятся на партиции. Каждому сообщению в партиции присваивается последовательный, неизменяемый offset. Потребители читают сообщения из партиции, поддерживая свою текущую позицию, также известную как их committed offset. Брокер Kafka отслеживает log-end-offset для каждой партиции, который представляет собой offset последнего сообщения, добавленного в нее.

Отставание потребителя = Log-End-Offset - Committed Offset

По сути, отставание — это количество сообщений, на которое потребитель отстает от головы лога для данной партиции. Хотя некоторое отставание естественно и ожидаемо в любой потоковой системе, постоянно растущее или чрезмерно большое отставание сигнализирует о проблеме.

Почему высокое отставание потребителя вызывает беспокойство:

• Задержка обработки данных: Ваши приложения могут обрабатывать данные слишком медленно, что влияет на аналитику в реальном времени или критические бизнес-операции.
• Истощение ресурсов: Потребители могут с трудом успевать, что приводит к высокой загрузке ЦП, памяти или сети.
• Устаревшие данные: Нижестоящие системы, получающие данные от отстающих потребителей, будут работать с устаревшей информацией.
• Проблемы с политикой хранения: Если отставание превышает период хранения топика, потребители могут навсегда пропустить сообщения, так как они будут удалены из лога.
• Ребалансировка группы потребителей: Постоянное отставание может способствовать нестабильному поведению группы потребителей и частым ребалансировкам.

Распространенные причины высокого отставания:

• Медленная логика потребителя: Само приложение-потребитель тратит слишком много времени на обработку каждого сообщения.
• Недостаточное количество экземпляров потребителей: Недостаточно запущенных экземпляров потребителей для обработки объема сообщений по всем партициям.
• Сетевая задержка: Проблемы между потребителями и брокерами.
• Проблемы производительности брокера: Брокеры могут испытывать трудности с эффективной доставкой сообщений.
• Всплески производства сообщений: Временные всплески сообщений, которые перегружают потребителей.
• Ошибки конфигурации: Неправильные конфигурации потребителя или топика.

Диагностика отставания с помощью kafka-consumer-groups.sh (рекомендуется)

Инструмент kafka-consumer-groups.sh — это современный и рекомендуемый способ управления и проверки групп потребителей. Он взаимодействует напрямую с брокерами Kafka для получения информации о смещениях потребителей, которая хранится во внутреннем топике __consumer_offsets. Этот инструмент предоставляет подробную информацию о состоянии группы потребителей, включая отставание.

Базовое использование для описания группы потребителей

Чтобы проверить отставание для конкретной группы потребителей, используйте опции --describe и --group:

kafka-consumer-groups.sh --bootstrap-server <Kafka_Broker_Host:Port> --describe --group <Consumer_Group_Name>

Замените <Kafka_Broker_Host:Port> на адрес одного из ваших брокеров Kafka (например, localhost:9092) и <Consumer_Group_Name> на имя группы потребителей, которую вы хотите проверить.

Интерпретация вывода

Типичный вывод будет выглядеть примерно так:

GROUP           TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                       HOST            CLIENT-ID
my-consumer-app my-topic                       0          12345           12347           2               consumer-1-a1b2c3d4-e5f6-7890-1234-abcdedfg      /192.168.1.100  consumer-1
my-consumer-app my-topic                       1          20000           20500           500             consumer-2-hijk-lmno-pqrs-tuvw-xyz              /192.168.1.101  consumer-2
my-consumer-app my-topic                       2          5000            5000            0               consumer-3-1234-5678-90ab-cdef-12345678          /192.168.1.102  consumer-3
my-consumer-app another-topic                  0          900             900             0               consumer-1-a1b2c3d4-e5f6-7890-1234-abcdedfg      /192.168.1.100  consumer-1

Давайте разберем важные столбцы:

• GROUP: Имя группы потребителей.
• TOPIC: Потребляемый топик.
• PARTITION: Конкретная партиция топика.
• CURRENT-OFFSET: Последний offset, зафиксированный потребителем для этой партиции.
• LOG-END-OFFSET: Offset последнего сообщения в этой партиции.
• LAG: Разница между LOG-END-OFFSET и CURRENT-OFFSET. Это количество сообщений, на которое потребитель отстает.
• CONSUMER-ID: Уникальный идентификатор экземпляра потребителя. Если это -, это означает, что этой партиции не назначен активный потребитель.
• HOST: IP-адрес или имя хоста экземпляра потребителя.
• CLIENT-ID: Идентификатор клиента, настроенный для экземпляра потребителя.

Ключевые наблюдения:

• Высокие значения LAG: Указывают на то, что потребитель отстает. Исследуйте логику потребителя, ресурсы или масштабирование.
• - в CONSUMER-ID: Предполагает, что партиция не потребляется. Это может быть связано с недостаточным количеством активных потребителей в группе или с тем, что экземпляр потребителя упал и не переподключился. Если LAG для таких партиций высок, это критическая проблема.
• LAG равный 0: Означает, что потребитель полностью догнал последние сообщения.

Диагностика отставания с помощью consumer-offset-checker.sh (устаревший инструмент)

consumer-offset-checker.sh — это более старый, устаревший инструмент, который полагался на ZooKeeper для хранения и получения смещений потребителей (для потребителей, использующих старый kafka.consumer.ZookeeperConsumerConnector). Для современных клиентов Kafka (0.9.0 и новее) смещения хранятся в самом Kafka. Хотя он в значительной степени заменен kafka-consumer-groups.sh, вы можете столкнуться с ним в старых средах или с устаревшими клиентами потребителей.

Предупреждение: Уведомление об устаревании

Этот инструмент полагается на ZooKeeper для управления смещениями. Современные клиенты Kafka (0.9.0+) хранят смещения непосредственно в Kafka. Для новых кластеров и клиентов kafka-consumer-groups.sh является авторитетным и предпочтительным инструментом. Используйте consumer-offset-checker.sh, только если вы точно знаете, что ваши клиенты-потребители настроены на хранение смещений в ZooKeeper.

Базовое использование

Чтобы проверить отставание с помощью этого инструмента, вам нужно указать строку подключения ZooKeeper:

consumer-offset-checker.sh --zk <ZooKeeper_Host:Port> --group <Consumer_Group_Name>

Замените <ZooKeeper_Host:Port> (например, localhost:2181) и <Consumer_Group_Name>.

Интерпретация вывода

Group           Topic                          Partition Offset  LogSize Lag     Owner
my-old-app      my-old-topic                   0         1000    1050    50      consumer-1_hostname-1234-5678-90ab-cdef
my-old-app      my-old-topic                   1         2000    2000    0       consumer-2_hostname-abcd-efgh-ijkl-mnop

• Group, Topic, Partition: Аналогично kafka-consumer-groups.sh.
• Offset: Зафиксированный offset потребителем.
• LogSize: LOG-END-OFFSET партиции.
• Lag: Количество сообщений, на которое потребитель отстает.
• Owner: Экземпляр потребителя, которому в настоящее время принадлежит (потребляет) партиция.

Интерпретация значений отставания аналогична: высокое отставание указывает на проблемы, а отсутствие Owner для партиции с высоким отставанием является критической проблемой.

Устранение высокого отставания потребителя: стратегии и сброс смещений

После того как вы определили высокое отставание потребителя, следующим шагом будет его устранение. Это часто включает двусторонний подход: во-первых, расследование и устранение первопричины, и во-вторых, при необходимости, сброс смещений потребителя.

Расследование первопричины

Прежде чем переходить к сбросу смещений, крайне важно понять, почему возникает отставание. Проверьте следующее:

• Логи приложения-потребителя: Ищите ошибки, чрезмерное время обработки или признаки сбоя приложения.
• Метрики хоста потребителя: Отслеживайте использование ЦП, памяти и сети. Ограничен ли потребитель по ресурсам?
• Метрики брокера Kafka: Находятся ли брокеры под нагрузкой? Высок ли дисковой ввод-вывод, сеть или ЦП?
• Пропускная способность продюсера: Был ли неожиданный всплеск производства сообщений?
• Состояние группы потребителей: Часто ли происходят ребалансировки? Достигается ли max.poll.interval.ms?

Масштабирование потребителей

Если проблема в том, что существующие потребители не могут обрабатывать сообщения достаточно быстро, и топик имеет достаточно партиций, вам может потребоваться масштабировать вашу группу потребителей, добавив больше экземпляров потребителей. Каждый экземпляр потребителя в группе возьмет на себя одну или несколько партиций, пока все партиции не будут назначены, вплоть до количества партиций.

Сброс смещений потребителя

Сброс смещений потребителя означает изменение начальной точки, с которой группа потребителей будет читать сообщения. Это мощная, потенциально разрушительная операция, которую следует использовать с осторожностью.

Важные соображения перед сбросом смещений:

• Потеря данных: Сброс на --to-latest приведет к тому, что потребители пропустят все сообщения между их текущим смещением и log-end-offset, что приведет к безвозвратной потере данных для этих сообщений.
• Повторная обработка данных: Сброс на --to-earliest или более старый offset означает, что потребители повторно обработают сообщения, которые они уже обработали. Ваше приложение-потребитель должно быть идемпотентным (обработка сообщения несколько раз дает тот же результат), чтобы корректно это обработать.
• Состояние приложения: Подумайте, как повторная обработка может повлиять на любое состояние, управляемое вашим приложением-потребителем или нижестоящими системами.

Чтобы сбросить смещения, вы снова будете использовать kafka-consumer-groups.sh. Он предлагает различные опции для сброса смещений:

• --to-earliest: Сбрасывает смещения на самый ранний доступный offset в партиции.
• --to-latest: Сбрасывает смещения на последний offset в партиции (фактически пропуская все текущие сообщения).
• --to-offset <offset>: Сбрасывает смещения на конкретный желаемый offset.
• --to-datetime <YYYY-MM-DDTHH:mm:SS.sss>: Сбрасывает смещения на offset, соответствующий определенной временной метке.
• --shift-by <N>: Сдвигает текущий offset на N позиций (например, -10, чтобы вернуться на 10 сообщений назад, +10, чтобы перейти на 10 сообщений вперед).

Критические функции безопасности: --dry-run и --execute

Всегда сначала выполняйте --dry-run, чтобы увидеть, что сделала бы операция сброса, прежде чем фиксировать ее с помощью --execute.

Пошаговый процесс сброса смещений:

1. Остановите всех потребителей в целевой группе потребителей. Это важно, чтобы предотвратить фиксацию новых смещений потребителями, пока вы пытаетесь их сбросить.

2. Выполните пробный запуск, чтобы просмотреть изменения смещений:

   • Пример: Сброс на самый ранний offset (повторная обработка всех сообщений)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-earliest --topic my-topic --dry-run
     

   • Пример: Сброс на последний offset (пропуск всех отставших сообщений)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-latest --topic my-topic --dry-run
     

   • Пример: Сброс на определенную временную метку (например, начать с 2023-01-01 00:00:00 UTC)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-datetime 2023-01-01T00:00:00.000 --topic my-topic --dry-run
     

   • Пример: Сдвиг смещений назад на 500 сообщений (на партицию)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --shift-by -500 --topic my-topic --dry-run
     

   Вывод --dry-run покажет предлагаемые изменения смещений:

   GROUP                          TOPIC                          PARTITION NEW-OFFSET
   my-consumer-app                my-topic                       0         0
   my-consumer-app                my-topic                       1         0
   

3. Выполните сброс, как только вы будете удовлетворены результатами пробного запуска:

   • Пример: Сброс на самый ранний offset (выполнение)

     kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-app --reset-offsets --to-earliest --topic my-topic --execute
     

4. Перезапустите приложения-потребители. После сброса смещений перезапустите экземпляры потребителей. Теперь они начнут потребление с новых начальных смещений.

Совет: Сброс для всех топиков в группе

Если вы хотите сбросить смещения для всех топиков, потребляемых группой, вы можете опустить флаг --topic при использовании kafka-consumer-groups.sh --reset-offsets. Будьте особенно осторожны с этим, так как это влияет на все.

Лучшие практики для операций с потребителями

• Проактивный мониторинг: Внедрите надежный мониторинг отставания потребителей с помощью таких инструментов, как Prometheus/Grafana, Datadog или пользовательских скриптов. Настройте оповещения для быстрорастущего или постоянно высокого отставания.
• Понимание идемпотентности: Проектируйте свои приложения-потребители так, чтобы они были идемпотентными. Это позволяет безопасно повторно обрабатывать сообщения в случае сбоев или сбросов смещений.
• Настройка max.poll.interval.ms: Этот параметр определяет максимальное время, в течение которого потребитель может не выполнять опрос. Если ваша логика обработки медленная, увеличьте это значение, чтобы предотвратить нежелательные ребалансировки, но также исследуйте основную причину медлительности.
• Обработка необрабатываемых сообщений: Реализуйте стратегию для сообщений "отравленная пилюля" (например, отправка их в очередь недоставленных сообщений - DLQ), а не повторные сбои и блокировка потребителя.
• Корректное завершение работы: Убедитесь, что ваши приложения-потребители корректно завершают работу, фиксируя свои последние смещения, чтобы избежать ненужной повторной обработки или скачков отставания во время перезапусков.
• Соответствие партиций потребителям: Для оптимального параллелизма стремитесь иметь как минимум столько же партиций, сколько вы планируете запускать экземпляров потребителей. Больше партиций обеспечивает больший параллелизм.

Практический поток инцидентов

Когда срабатывает оповещение об отставании, удержитесь от желания сначала сбросить смещения. Начните с захвата текущего состояния группы:

kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group payments-writer

Смотрите на форму, а не только на размер. Если каждая партиция отстает примерно на одинаковую величину, вся группа, вероятно, недообеспечена ресурсами или заблокирована общей зависимостью. Если одна партиция сильно отстает, проверьте перекос ключей, сообщение-отраву или один хост потребителя с плохим поведением ЦП, диска, DNS или сети. Если CONSUMER-ID равно -, у партиции в данный момент нет назначенного активного члена; это обычно указывает на упавших потребителей, текущую ребалансировку или группу с меньшим количеством здоровых членов, чем ожидалось.

Запустите команду снова через минуту. Значение отставания 500 000 менее тревожно, если оно быстро падает после отката развертывания. Значение отставания 5 000 более тревожно, если оно удваивается каждую минуту во время нормального трафика. Во время инцидента я обычно записываю три числа: общее отставание, отставание худшей партиции и стабильно ли состояние группы. Это дает вам достаточно сигналов, чтобы решить, масштабировать ли потребителей, замедлить продюсеров, исправить ошибки приложения или подготовить контролируемое воспроизведение.

Перед любым сбросом сохраните текущие смещения в надежном месте, даже если это просто тикет инцидента. Пробный запуск — это не резервная копия. Вывод команды дает вам смещения, которые могут понадобиться, если кто-то поймет, что сброс пропустил данные, которые все еще были важны.

Финальные проверки

Здоровый runbook по отставанию имеет три привычки: описывать перед изменением, выполнять пробный запуск перед выполнением и исправлять потребителя перед перемещением смещений. kafka-consumer-groups.sh дает вам необработанную правду о зафиксированных смещениях и владении партициями. Ваша задача — связать этот вывод с поведением приложения, стоящим за ним.