Решение проблем Apache Kafka

Симптомы

• Producer или consumer получает NotLeaderOrFollowerException при попытке записи или чтения
• Количество повторных попыток резко возрастает в метриках клиента
• Ошибки носят временный характер и проходят после обновления метаданных
• В логах брокера видны сообщения о смене лидера партиции

Причина

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

Решение

1. Увеличьте metadata.max.age.ms (например, до 60000 мс), чтобы клиент чаще обновлял метаданные
2. Настройте retries >= 10 и retry.backoff.ms >= 200 для автоматического повтора после обновления метаданных
3. Проверьте состояние брокеров командой kafka-topics.sh --describe и убедитесь в наличии лидера для каждой партиции
4. Мониторьте UncleanLeaderElectionsPerSec — ненулевое значение указывает на проблемы с репликацией

Связанные уроки:

Симптомы

• Консьюмеры прекращают обработку сообщений, лаг партиций растёт
• В логах постоянно появляются записи 'Attempt to heartbeat failed since group is rebalancing'
• Повторные вызовы commitSync() или commitAsync() завершаются ошибкой CommitFailedException
• Консьюмер-группа не достигает стабильного состояния — ребалансировка запускается снова и снова

Причина

Консьюмер превысил session.timeout.ms (не отправил heartbeat вовремя) или max.poll.interval.ms (слишком долго обрабатывал записи между вызовами poll()). Это интерпретируется координатором группы как сбой участника и запускает ребалансировку.

Решение

1. Увеличьте max.poll.interval.ms до значения, превышающего максимальное время обработки одного батча
2. Уменьшите max.poll.records, чтобы сократить объём работы между вызовами poll()
3. Перейдите на CooperativeStickyAssignor — он минимизирует число партиций, передаваемых при ребалансировке
4. Рассмотрите использование статического членства (group.instance.id) для консьюмеров с предсказуемым временем перезапуска

Связанные уроки:

Симптомы

• Продюсер не может отправить сообщения — все попытки завершаются LeaderNotAvailableException
• Операции с топиком (describe, produce, consume) временно недоступны
• Ошибка возникает сразу после создания нового топика или при запуске кластера
• В логах брокера видны записи о выборах лидера (leader election)

Причина

Лидер партиции ещё не избран — это типично для только что созданного топика или после того, как ISR опустел и unclean.leader.election.enable=false. Брокер возвращает ошибку, пока контроллер не завершит выборы.

Решение

1. Подождите несколько секунд: выборы лидера в KRaft обычно занимают менее 1 секунды, но при нестабильном кластере могут затягиваться
2. Проверьте состояние ISR: kafka-topics.sh --describe должен показывать непустой ISR для каждой партиции
3. Убедитесь, что replication.factor не превышает число доступных брокеров
4. Если ошибка носит постоянный характер, проверьте логи контроллера на наличие ошибок выборов

Связанные уроки:

Симптомы

• Продюсер с настройкой acks=all получает NotEnoughReplicasException и не может записать сообщения
• Запись в топик полностью блокируется до восстановления достаточного числа реплик
• В метриках брокера значение UnderReplicatedPartitions значительно больше нуля
• Один или несколько брокеров отстают от лидера или полностью недоступны

Причина

Число живых синхронизированных реплик (ISR) стало меньше min.insync.replicas. Это происходит, когда брокеры падают, сетевая связность между брокерами нарушена, или фолловер не успевает за лидером и вылетает из ISR по таймауту replica.lag.time.max.ms.

Решение

1. Верните упавшие брокеры в строй — это восстановит ISR и позволит возобновить запись
2. При длительном простое временно уменьшите min.insync.replicas до 1 на критических топиках (с пониманием рисков)
3. Проверьте сетевую связность между брокерами: высокая задержка или потери пакетов выбрасывают реплики из ISR
4. Увеличьте replica.lag.time.max.ms, если фолловеры отстают из-за нагрузки дискового I/O

Связанные уроки:

Симптомы

• Консьюмер неожиданно сбрасывается на самое раннее или самое позднее смещение
• В логах появляется OffsetOutOfRangeException при попытке получить данные
• Часть сообщений пропускается или обрабатывается повторно после сброса смещения
• Проблема возникает после длительного простоя консьюмер-группы

Причина

Зафиксированное смещение больше не существует в топике: данные были удалены политикой хранения (retention.ms истёк или диск заполнен), либо зафиксированное смещение указывает за пределы текущего конца лога (LEO). Чаще всего это происходит, когда консьюмер-группа долго не работала и offsets.retention.minutes истёк.

Решение

1. Установите auto.offset.reset=earliest, чтобы при потере смещения начинать с доступных данных
2. Увеличьте offsets.retention.minutes на брокере (по умолчанию 7 дней) для хранения смещений спящих групп
3. Настройте мониторинг consumer lag — рост лага сигнализирует о проблемах задолго до потери смещений
4. Используйте kafka-consumer-groups.sh --reset-offsets для ручного задания смещения при необходимости

Связанные уроки:

Симптомы

• Продюсер отклоняет сообщения с RecordTooLargeException ещё на стороне клиента
• Брокер возвращает ошибку MESSAGE_TOO_LARGE при попытке записи
• Проблема стабильно воспроизводится для сообщений определённого размера
• Компрессия включена, но сообщения всё равно превышают лимит

Причина

Размер сообщения превышает один из трёх лимитов: max.request.size на продюсере (по умолчанию 1 МБ), message.max.bytes на брокере (по умолчанию 1 МБ) или max.message.bytes на уровне топика. Все три значения должны быть согласованы — несоответствие хотя бы одного вызывает ошибку.

Решение

1. Согласуйте все три параметра: max.request.size (продюсер) >= message.max.bytes (брокер) >= max.message.bytes (топик)
2. Включите компрессию на продюсере: compression.type=lz4 или snappy снижает размер батча в 3-5 раз
3. Рассмотрите разбиение больших сообщений на части с последующей сборкой на стороне консьюмера
4. Для бинарных данных используйте внешнее хранилище (S3, HDFS) и передавайте через Kafka только ссылки

Связанные уроки:

Симптомы

• Продюсер сообщает о TimeoutException, батчи сообщений истекают не отправленными
• Метрика record-error-rate растёт, throughput падает
• Проблема возникает при пиковой нагрузке или при замедлении брокеров
• Очередь буфера (buffer.memory) постоянно заполнена — новые записи блокируются

Причина

Брокер не подтверждает запросы за время delivery.timeout.ms (по умолчанию 120 000 мс). Причины: брокер перегружен медленным дисковым I/O или GC-паузами, сетевая задержка слишком высокая, или buffer.memory переполнен из-за высокого темпа продукции.

Решение

1. Увеличьте delivery.timeout.ms пропорционально реальным задержкам подтверждений в p99
2. Проверьте нагрузку на брокеры: метрики RequestHandlerAvgIdlePercent и NetworkProcessorAvgIdlePercent должны быть выше 30%
3. Увеличьте buffer.memory (по умолчанию 32 МБ) при высоком темпе продукции
4. Настройте compression.type=lz4 для уменьшения объёма данных, передаваемых по сети

Связанные уроки:

Симптомы

• Вызов commitSync() бросает CommitFailedException после завершения обработки батча
• Консьюмер теряет назначенные партиции в процессе обработки
• Одни и те же сообщения обрабатываются несколько раз разными экземплярами
• В логах видны записи о смене поколения группы (generation changed)

Причина

Ребалансировка завершилась, пока консьюмер обрабатывал записи — координатор присвоил новое поколение группы. Попытка зафиксировать смещения для старого поколения отклоняется. Корневая причина: обработка батча заняла больше времени, чем max.poll.interval.ms.

Решение

1. Используйте commitAsync() с обратным вызовом для асинхронной фиксации — фиксация не блокирует обработку
2. Увеличьте max.poll.interval.ms до значения, в 1.5-2 раза превышающего максимальное время обработки батча
3. Уменьшите max.poll.records, чтобы каждый батч обрабатывался значительно быстрее таймаута
4. Перенесите тяжёлые операции (вызовы внешних API, запись в БД) в асинхронные задачи с отдельным пулом потоков

Связанные уроки:

Симптомы

• Продюсер или консьюмер получает UnknownTopicOrPartitionException при первом обращении
• Ошибка стабильная, а не временная — топик не появляется после повторных попыток
• Метаданные не содержат информации о запрашиваемом топике
• Операции с другими топиками работают нормально

Причина

Топик не существует в кластере. Возможные причины: auto.create.topics.enable=false на брокере (типично для production), опечатка в имени топика, или топик не был создан в рамках процесса деплоя. В production-кластерах автосоздание топиков отключают намеренно.

Решение

1. Создайте топик заранее командой: kafka-topics.sh --create --topic --partitions --replication-factor
2. Дважды проверьте имя топика — Kafka чувствительна к регистру и не допускает специальных символов кроме точек, дефисов и подчёркиваний
3. В CI/CD пайплайне добавьте шаг создания топиков перед деплоем приложения
4. Если auto.create.topics.enable=true и ошибка всё равно возникает, проверьте права доступа (ACL)

Связанные уроки:

Симптомы

• Клиенты не могут подключиться к брокеру — соединение разрывается на этапе SSL-рукопожатия
• В логах брокера появляются записи 'SSL handshake failed' или 'certificate_unknown'
• Проблема воспроизводится для всех клиентов, а не только для одного
• Ошибка может возникать периодически — незадолго до или после истечения срока сертификата

Причина

Несоответствие SSL-конфигурации: клиент не доверяет сертификату брокера (отсутствует CA в truststore), сертификат истёк, имя хоста в сертификате не совпадает с адресом брокера, или между клиентом и брокером несовместимые версии TLS-протокола.

Решение

1. Проверьте срок действия сертификатов: openssl s_client -connect broker:9093 -showcerts покажет цепочку и дату истечения
2. Убедитесь, что CA-сертификат брокера добавлен в truststore всех клиентов
3. При ошибке hostname: либо добавьте правильный SAN в сертификат, либо установите ssl.endpoint.identification.algorithm='' (только для dev)
4. Проверьте совместимость TLS-версий: брокер и клиент должны поддерживать общие cipher suites

Связанные уроки:

Симптомы

• Коннектор не запускается — воркер сообщает об ошибке поиска класса коннектора
• REST API Kafka Connect возвращает 500 при попытке создать коннектор
• Другие коннекторы, установленные ранее, работают без проблем
• Ошибка воспроизводится при перезапуске воркера

Причина

JAR-файл коннектора отсутствует в директориях, указанных в plugin.path воркера, или класс коннектора не находится в пути поиска плагинов. Kafka Connect использует изолированные classloaders для каждого плагина — JAR должен быть в отдельной поддиректории plugin.path.

Решение

1. Убедитесь, что JAR коннектора и все его зависимости находятся в отдельной папке внутри plugin.path
2. Проверьте значение plugin.path в connect-distributed.properties — укажите несколько путей через запятую при необходимости
3. После добавления JAR перезапустите воркер — плагины сканируются только при старте
4. Используйте confluent-hub install или ручную установку по документации конкретного коннектора

Связанные уроки:

Симптомы

• Sink-коннектор останавливается с ошибкой, статус задачи переходит в FAILED
• В логах воркера виден стектрейс SerializationException или SchemaException
• Проблема возникает при изменении схемы сообщений в исходном топике
• Другие задачи того же коннектора продолжают работать

Причина

Sink-задача не может десериализовать входящие сообщения из-за несоответствия конвертера (value.converter) формату данных в топике. Типичные причины: схема изменилась и несовместима с текущей версией в Schema Registry, URL Schema Registry не задан, или сообщения в топике имеют смешанный формат.

Решение

1. Проверьте и согласуйте настройки key.converter и value.converter с форматом данных в топике
2. Задайте schema.registry.url в конфигурации воркера и коннектора
3. Настройте Dead Letter Queue (errors.deadletterqueue.topic.name) для изоляции проблемных сообщений без остановки задачи
4. Включите errors.tolerance=all и errors.log.enable=true для диагностики без прерывания обработки

Связанные уроки:

Симптомы

• Приложение Kafka Streams не запускается — выбрасывается StreamsException при старте
• Исключение содержит список топиков, которые не найдены в кластере
• Приложение работало нормально в dev-среде, но не запускается в staging или production
• Другие приложения, использующие те же топики, работают корректно

Причина

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

Решение

1. Создайте входные топики перед запуском приложения — добавьте шаг создания топиков в процесс деплоя
2. Вызовите topology.describe() в логировании при старте для проверки всех топиков, на которые ссылается топология
3. Сравните имена топиков в конфигурации приложения и в кластере с учётом среды (dev/staging/prod)
4. Настройте StreamsConfig.REPLICATION_FACTOR_CONFIG — Streams создаёт changelog и repartition топики автоматически

Связанные уроки:

Симптомы

• Создание топика завершается ошибкой InvalidReplicationFactorException
• Ошибка возникает при выполнении kafka-topics.sh --create или при автосоздании топика
• В кластере работает меньше брокеров, чем указано в replication.factor
• Другие топики с меньшим replication.factor создаются без проблем

Причина

Значение replication.factor при создании топика превышает количество доступных (живых) брокеров в кластере. Kafka не может разместить реплики на несуществующих брокерах, поэтому отклоняет запрос на создание топика.

Решение

1. Убедитесь, что число живых брокеров не меньше replication.factor: kafka-broker-api-versions.sh покажет доступные брокеры
2. Для dev-окружений используйте replication.factor=1 и min.insync.replicas=1 — это нормально без требований к отказоустойчивости
3. Настройте default.replication.factor на брокере для автосоздаваемых топиков
4. При горизонтальном масштабировании сначала добавьте брокеры, потом увеличивайте replication.factor

Связанные уроки:

Симптомы

• poll() возвращает пустые записи снова и снова, хотя в топике есть непрочитанные сообщения
• ID поколения группы в логах постоянно меняется — ребалансировки следуют одна за другой
• Некоторые экземпляры консьюмеров не получают ни одной партиции
• kafka-consumer-groups.sh --describe показывает консьюмеров с нулевым числом партиций

Причина

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

Решение

1. Проверьте здоровье каждого экземпляра консьюмера — найдите зависающие или медленные обработчики
2. Включите статическое членство группы через group.instance.id — это позволяет кратковременно пропадающим консьюмерам переподключаться без ребалансировки
3. Убедитесь в отсутствии zombie-консьюмеров: процессы, которые числятся в группе, но уже не работают
4. Перейдите на CooperativeStickyAssignor для инкрементальной ребалансировки без остановки всей обработки

Связанные уроки:

Симптомы

• Брокер внезапно завершает работу с OutOfMemoryError в логах или в hs_err_pid файле
• Производительность брокера деградирует перед падением: GC-паузы растут, обработка запросов замедляется
• OOM возникает при высоком числе одновременных соединений или при больших fetch-запросах
• После перезапуска брокер снова падает под той же нагрузкой

Причина

Хип JVM брокера исчерпан. Основные причины: слишком маленький -Xmx для текущей нагрузки, большое число соединений (каждое требует буферов), слишком большой fetch.max.bytes в запросах консьюмеров, или утечка памяти в обработчиках запросов.

Решение

1. Увеличьте размер хипа через KAFKA_HEAP_OPTS: -Xmx6g -Xms6g (типично 4-8 ГБ для production)
2. Настройте max.connections.per.ip и max.connections для ограничения числа одновременных подключений
3. Ограничьте fetch.max.bytes и max.partition.fetch.bytes в конфигурации консьюмеров
4. Включите JVM GC-логирование (-Xlog:gc*) и анализируйте паттерн выделения памяти перед OOM

Связанные уроки:

Симптомы

• Клиенты не могут пройти аутентификацию, в логах видна ошибка поиска LoginModule
• Ошибка возникает сразу при подключении до отправки каких-либо данных
• В конфигурации SASL указан механизм (PLAIN, SCRAM, GSSAPI), но соединение не устанавливается
• Брокер в логах показывает 'Failed authentication' для всех подключающихся клиентов

Причина

SASL настроен в security.protocol, но JVM не может найти класс LoginModule, указанный в JAAS-конфигурации. Типичные причины: JAAS-файл не передан JVM через параметр -Djava.security.auth.login.config, имена секций KafkaServer или KafkaClient написаны с ошибкой, или нужный модуль (krb5, scram) отсутствует в classpath.

Решение

1. Передайте JAAS-конфигурацию через JVM-аргумент: -Djava.security.auth.login.config=/path/to/kafka_jaas.conf
2. Проверьте точность имён секций в JAAS: для брокера — KafkaServer, для клиента — KafkaClient (чувствительно к регистру)
3. Для SASL/PLAIN и SCRAM убедитесь, что PlainLoginModule или ScramLoginModule присутствует в classpath Kafka
4. Для Kerberos (GSSAPI) проверьте наличие krb5.conf и корректность principal/keytab

Связанные уроки:

Симптомы

• Метрика UnderReplicatedPartitions ненулевая длительное время — более 30 секунд
• Один или несколько брокеров значительно отстают от лидеров партиций
• Производительность записи падает при acks=all — продюсеры ждут подтверждений от отстающих реплик
• Брокер потребляет аномально высокий дисковый I/O или сетевой трафик

Причина

Фолловеры не успевают догнать лидера и вылетают из ISR. Причины: перегруженность диска на брокере-фолловере, высокая сетевая задержка между брокерами, неравномерное распределение лидеров (один брокер является лидером для слишком многих партиций), или временный сбой одного брокера.

Решение

1. Проверьте дисковый I/O на отстающих брокерах: iostat -x 1 покажет утилизацию дисков
2. Перебалансируйте лидеров партиций: kafka-leader-election.sh для равномерного распределения нагрузки
3. Увеличьте replica.fetch.max.bytes и num.replica.fetchers на отстающем брокере для ускорения репликации
4. Проверьте сетевую связность между брокерами и увеличьте replica.lag.time.max.ms при высоких задержках

Связанные уроки:

Симптомы

• Брокер не может запуститься — в логах видна ошибка CRC при загрузке сегмента лога
• Партиция становится недоступной после аварийного отключения питания или unclean shutdown
• kafka-dump-log.sh показывает повреждённые записи в конкретном сегменте
• Брокер запускается, но партиция остаётся в состоянии offline

Причина

Файл сегмента лога повреждён — контрольная сумма CRC записей не совпадает. Чаще всего это следствие аварийного отключения без flush данных на диск, аппаратного сбоя (RAID без записи, битый сектор), или файловой системы без journaling.

Решение

1. Определите повреждённый сегмент командой: kafka-dump-log.sh --files /path/to/segment.log --print-data-log 2>&1 | grep -i corruption
2. Если кластер работает с репликацией, удалите повреждённые сегменты на проблемном брокере и дайте репликации восстановить данные
3. При отсутствии реплик: удалите повреждённый сегмент вручную и перезапустите брокер — данные из сегмента будут потеряны
4. Для профилактики используйте файловую систему ext4 или xfs с journaling и настройте UPS для брокеров

Связанные уроки:

Симптомы

• Транзакционный продюсер получает ProducerFencedException при вызове beginTransaction()
• Все незавершённые транзакции текущего экземпляра продюсера прерываются
• Проблема возникает при перезапуске приложения или при запуске нескольких экземпляров
• Консьюмеры с isolation.level=read_committed не получают сообщения из прерванных транзакций

Причина

Два экземпляра продюсера используют одинаковый transactional.id, или новый экземпляр был создан раньше, чем старый завершил свои транзакции. Kafka повышает эпоху продюсера (producer epoch) — старые экземпляры с меньшей эпохой блокируются для предотвращения дублирования сообщений.

Решение

1. Гарантируйте, что только один активный экземпляр использует данный transactional.id — реализуйте лидерную блокировку (leader election) при горизонтальном масштабировании
2. Перехватывайте ProducerFencedException отдельно от других исключений — это сигнал о создании нового экземпляра продюсера, а не о повторной попытке
3. При перезапуске приложения создавайте новый экземпляр KafkaProducer — старый экземпляр не подлежит восстановлению
4. На стороне консьюмеров установите isolation.level=read_committed для чтения только завершённых транзакций

Связанные уроки:

Симптомы

• Регистрация новой версии схемы в Schema Registry завершается ошибкой совместимости
• Продюсер не может отправить сообщения с обновлённой схемой
• В логах видна ошибка RestClientException с указанием нарушенного правила совместимости
• Другие схемы в Registry продолжают работать нормально

Причина

Новая версия схемы нарушает режим совместимости, установленный для данного subject в Schema Registry. При BACKWARD-совместимости консьюмеры со старой схемой должны уметь читать данные, записанные с новой. Удаление обязательного поля или изменение типа без значения по умолчанию нарушает это правило.

Решение

1. Изучите режим совместимости топика: GET /config/ через Schema Registry REST API
2. При добавлении полей указывайте значение по умолчанию (default) — это обеспечивает BACKWARD-совместимость
3. Никогда не удаляйте поля из схемы при BACKWARD или FULL совместимости — помечайте их как deprecated
4. При необходимости кардинального изменения схемы: используйте новый subject или TopicNameStrategy с версионированием

Связанные уроки: