Kafka consumer застрял на одном событии: poison pill и DLQ

Consumer каждый раз падает на offset 42. После перезапуска он снова читает ту же запись и снова падает, поэтому offset 43 и следующие не обрабатываются. Такая запись получила название poison pill: для текущего кода её нельзя разобрать или применить, а повтор без изменения условий не помогает.

Первое действие — определить вид ошибки. Таймаут внешней БД может исчезнуть после восстановления сервиса; отправлять корректную запись в DLQ из-за одной сетевой паузы рано. Если payload необратимо нарушает контракт, например вместо ожидаемого числа содержит строку, повтор той же версии кода снова упадёт. Нужен путь для изоляции записи и продвижения потока.

Где именно возникает ошибка

Если используемый Kafka deserializer выбрасывает исключение внутри poll(), приложение может даже не получить обычный ConsumerRecord. В Java-клиенте RecordDeserializationException сообщает партицию и offset, а также даёт доступ к исходным буферам ключа и значения. Другой подход — получать байты через ByteArrayDeserializer, а разбор формата делать после poll() в коде приложения. Тогда проблемная запись становится обычной записью, которую можно маршрутизировать.

Для offset 42 из партиции 1 положите в DLQ исходные байты и служебные поля: исходные topic, partition, offset, тип ошибки и версию схемы. Идентификатор topic/partition/offset позволяет узнать повтор той же записи в DLQ. Не вставляйте бездумно весь payload в текстовый лог: он может содержать персональные данные; доступ к DLQ должен соответствовать доступу к исходной теме.

Порядок, который сохраняет след

Сначала получите подтверждение, что запись сохранена в DLQ или другом карантине. Затем зафиксируйте для исходной партиции позицию следующей записи и продолжите чтение. Если продвинуть offset до подтверждения DLQ, падение между действиями приведёт к потере проблемного события. Если сначала сохранить DLQ и упасть до commit offset, запись может попасть туда второй раз. Поэтому DLQ-потребитель должен распознавать повтор по исходному идентификатору.

Нельзя просто выполнять seek(43) и считать проблему решённой: seek меняет текущую позицию чтения, а после нового перезапуска группа начнёт с последнего сохранённого offset. Если принято решение пропустить запись, сохраните следующую позицию для соответствующей партиции после надёжного карантина. При пачке записей не продвигайте offset мимо других необработанных записей той же партиции.

DLQ — очередь работы, а не корзина. Для неё нужны счётчик поступлений, владелец разбора и правило повторной обработки после исправления схемы или данных. Иначе pipeline выглядит «зелёным», но часть бизнес-событий навсегда исключена из витрины. Полезно сверять количество полученных, успешно обработанных и отправленных в карантин сообщений за период.

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

Повторная обработка из DLQ требует правила порядка. Пока запись 42 лежала в карантине, основной поток мог обработать 43 и 44. Если после исправления просто опубликовать 42 в конец темы, для той же сущности она окажется после более поздних переходов. Обработчик должен уметь отклонять устаревшую версию события, компенсировать состояние либо остановить зависимую партицию до исправления. Выбор зависит от того, допускает ли бизнес пропуск одного события.

При ошибке разбора проверьте сырой payload и версию схемы, но не редактируйте сообщение в журнале. Обычно исправляют producer или consumer и переигрывают сохранённые байты из карантина с явной записью исходного идентификатора. Если вручную изменить значение, важно сохранить, кто и почему сделал исправление: иначе повторное расследование не сможет воспроизвести исходный сбой.

В курсе Data Engineering Fast‑Track модели доставки и группы consumer связываются с конвейером Kafka → Spark → lakehouse, схемами сообщений и проверкой свежести. Откройте программу, чтобы разбирать такие остановки в контексте всего потока данных.

После изоляции ошибки проверьте и порядок обработки и commit offset: он определяет, увидите ли вы запись повторно после сбоя.