Ошибки консьюмера и яд-сообщения

В двенадцатом уроке ошибки продюсера ретраились сами: Writer повторял пачку до MaxAttempts с бэкоффом, и код вызывающего получал уже итог. У консьюмера такой машины нет — ни в kafka-go, ни в протоколе Kafka: читатель вытащил сообщение, хендлер упал, и библиотека честно ждёт следующего решения, не делая никаких предположений о том, каким оно должно быть. Эта свобода мгновенно превращается в ловушку, как только в партицию попадает яд-сообщение: запись с битым JSON, несовместимой схемой или полем, которое валит код в панике. Хендлер падает, цикл возвращает ошибку наружу, процесс рестартует, группа перечитывает последний незакоммиченный офсет — и снова падает. Диспетчер асинхронных задач из курса asynq в такой ситуации посчитал бы попытки и передал бы задачу в архив; консьюмер Kafka без вашей помощи будет падать вечно, а вместе с ним встанет вся партиция. Этот урок — про то, как разорвать цикл осознанно: отличать ошибки, которые пройдут сами, от ошибок, которые не пройдут никогда, и уводить вторые в dead letter queue без потери остальных событий.

Цикл смерти: что делает яд с группой#

Механику цикла стоит прогнать по шагам один раз, чтобы узнавать её в продакшене с первых секунд. Шаг первый: FetchMessage возвращает сообщение с офсетом N — сам по себе он ничего не фиксирует, офсет по-прежнему указывает на N. Шаг второй: хендлер падает на невалидных данных, цикл возвращает ошибку и не доходит до CommitMessages — позиция группы осталась на N. Шаг третий: процесс рестартует или просто продолжит читать с того же места — FetchMessage снова возвращает то же сообщение с офсетом N, ведь смещение не двигалось. Дальше всё повторяется: партиция стоит, лаг растёт, а сообщения за яд-записью, вполне здоровые, копятся и стареют по ретеншену из семнадцатого урока. Длительный цикл опасен и для группы: читатель перестаёт укладываться в heartbeat-интервалы, срабатывает сессия, включается ребаланс, партиция перезжает на другого участника — и цикл смерти переезжает вместе с ней, потому что яд-сообщение по-прежнему на месте. Диагноз простой: один и тот же офсет в логах снова и снова — значит, перед вами poison message, а не случайный сбой.

Ошибки инфраструктуры против ошибок данных#

Первая линия обороны — классификация: перед тем как что-то предпринимать, спросите, способна ли эта ошибка исчезнуть сама. Ошибки инфраструктуры могут: сеть мигнула, брокер перезапускается, группа ушла в ребаланс — ретрай с бэкоффом здесь правильная реакция, и достаточно обычного цикла с паузой, потому что те же байты при следующей попытке прочитаются и обработаются нормально. Ошибки данных не исчезнут никогда: тот же JSON распарсится тем же парсером с тем же результатом хоть тысячу раз подряд, потому что ошибка сидит не в окружении, а в самом сообщении. Ретрай данных — это и есть цикл смерти, поэтому такие записи надо снимать с конвейера. Виды поведения складываются в три ветки: ошибка прошла — коммит и дальше; ошибка может пройти сама — ретрай без продвижения офсета; ошибка никогда не пройдёт — после порога попыток отправить сообщение в DLQ и закоммитить. Размытие границ между ветками стоит дорого: транспортный сбой, отправленный в DLQ, теряет событие, которое само бы дообработалось, а яд, отправленный на ретрай, останавливает партицию.

Ошибка → действие: шпаргалка цикла чтения#

Ошибка	Класс	Действие
Сетевой сбой, брокер недоступен	Инфраструктура	Ретрай с бэкоффом; офсет не трогать
io.EOF от FetchMessage	Штатный исход	Reader закрыт: выходим из цикла
context.Canceled при остановке	Штатный исход	Завершаемся; недообработанное перечитается
json.Unmarshal не прошёл	Данные	Счётчик неудач; после N — в DLQ и коммит
Неизвестная версия схемы	Данные	Сразу в DLQ: ретраи не изменят байты
Паника в хендлере	Данные либо код	recover в обёртке; дальше как ошибку данных

DLQ-паттерн: N неудач и в сторону#

Dead letter queue — отдельный топик, по имени обычно <topic>.dlq, куда уходит сырое проблемное сообщение целиком: значение как есть, ключ оригинальный, причина в заголовке. Порядок работы с порогом неудач: хендлер падает на ошибке данных, счётчик по ключу или по офсету увеличивается, пока порог не достигнут, офсет остаётся на месте — вдруг это всё-таки инфраструктура и следующая попытка пройдёт; как только порог исчерпан, сообщение целиком уходит в DLQ через Writer из четвёртого урока, и после успешной отправки цикл коммитит яд-запись и продолжается со следующей. Коммит здесь — ключевой жест: он перешагивает сообщение, цикл прерывается, партиция оживает, а яд не потерян — он лежит в отдельном топике и ждёт ручного разбора. В заголовок кладут причину и след ошибки: без них разбор DLQ превращается в археологию, потому что сырое значение само не расскажет, чем именно оно мешало хендлеру. Порог N обычно мал — две-три попытки достаточно, чтобы отличить мигнувшую сеть от настоящих данных, которые не поправятся.

Хендлер со счётчиком неудач и DLQ

const maxFails = 3

func (c *Consumer) handle(ctx context.Context, m kafka.Message) error {
        var order Order
        if err := json.Unmarshal(m.Value, &order); err == nil {
                return c.process(ctx, order) // данные валидны
        }

        c.fails[m.Key]++ // счётчик неудач по ключу

        if c.fails[m.Key] < maxFails {
                return errRetryLater // офсет на месте, попробуем позже
        }

        if err := c.dlq.WriteMessages(ctx, kafka.Message{
                Key:   m.Key,
                Value: m.Value, // сырое сообщение, без пересборки
                Headers: []kafka.Header{
                        {Key: "dlq-reason", Value: []byte("json")},
                },
        }); err != nil {
                return err // DLQ недоступна: не теряем сообщение
        }
        delete(c.fails, m.Key)
        return nil // яд снят с конвейера: коммит двинет офсет
}

Валидные данные обрабатываются обычным путём; ошибка unmarshal попадает в счётчик, а не сразу в DLQ — короткий порог отфильтрует инфраструктурные совпадения. В DLQ уходит сырое значение: пересобранное сообщение скрывает, чем именно был плох оригинал. Ошибка записи в DLQ не глотается: пока яд не лежит в очереди, коммитить нельзя, иначе событие потеряется навсегда.

Проверка: цикл без DLQ-логики

Консьюмер падает на битом JSON и возвращается к чтению без коммита. Что произойдёт дальше?

Счётчик в памяти против retry-топика#

Счётчик неудач в map внутри процесса — самое дешёвое решение, и для большинства сервисов достаточное: ноль сетевых ходов, ноль новых топиков, понятный код. У цены две строки в чеке: память обнуляется при рестарте — яд-сообщение снова пройдёт свой цикл неудач с нуля; и память обнуляется при ребалансе, если партиция уехала на другого участника, у которого счётчик пуст. Второй вариант — retry-топик: неудачная обработка переотправляет сообщение в <topic>.retry, отдельный читатель берёт его оттуда с задержкой, а после исчерпания попыток уже переносит в DLQ. Такой конвейер переживёт рестарты и ребалансы, даёт честную задержку между попытками — время на то, чтобы внешняя система, из-за которой падал хендлер, поправилась, — но требует инфраструктуры: двух топиков, двух читателей, аккуратных заголовков с числом попыток. Разумная стратегия старта: счётчик в памяти плюс DLQ, а retry-топик добавлять тогда, когда хендлер зависит от медленных внешних систем и короткий ретрай внутри процесса перестал помогать.

Живое место: обнулённый счётчик после ребаланса

Инцидент: сервис обрабатывал партиции тремя экземплярами, яд-сообщение болталось между подами при каждом ребалансе — счётчик у нового владельца пуст, и запись снова дважды падала и только на третьей попытке уходила в DLQ. Заметили это не по алертам, а по лишним ошибкам в логах. Причина честная: kafka-go не транслирует события ребаланса в ваш код автоматически, счётчики никто не переносил. Лечение по возрастанию сложности: считать неудачи по офсету сообщения, а не по ключу в глобальной map, — ошибка повторяется на одном и том же месте; либо сделать порог идемпотентным к повторам — первый же ход в DLQ безопасен для битого JSON, если он и вправду битый; либо перейти на retry-топик, где число попыток хранится в самом сообщении и переживает любые переезды.

Разбор DLQ отдельным читателем

dlqReader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"broker-1:9092"},
        Topic:   "orders.dlq",
        GroupID: "dlq-investigators",
})

for {
        m, err := dlqReader.FetchMessage(ctx)
        if err != nil {
                break // io.EOF или отмена ctx
        }

        log.Printf("DLQ %s key=%s reason=%s",
                m.Topic, m.Key, headerValue(m, "dlq-reason"))
        // ручной разбор: лог, алерт, перенос в исправленный топик

        if err := dlqReader.CommitMessages(ctx, m); err != nil {
                break
        }
}

DLQ — обычный топик, поэтому читается обычным Reader со своей группой: ни скорость, ни коммиты разборщика не мешают основному конвейеру. Заголовок dlq-reason, который хендлер положил при отправке, подсказывает класс проблемы без пересбора байтов. Типичный разборщик не восстанавливает данные автоматически — он логирует, алертит и держит запись до решения человека.

Проверка: коммит после DLQ

Сообщение успешно отправлено в orders.dlq. Почему цикл обязан закоммитить оригинал?

Что сделает корректный код при падении записи в DLQ?

Прочитайте код и выберите, что он выведет:

// хендлер: счётчик исчерпан, яд отправляется в DLQ
err := c.dlq.WriteMessages(ctx, kafka.Message{
        Key:   m.Key,
        Value: m.Value,
})
// но брокер DLQ-топика недоступен: err != nil

Гигиена DLQ: разбор, ретеншен, алерты#

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

Про песочницу: яд-сообщение на локальном стенде

Интерактивы урока исполняются в песочнице курса без сети, но весь конвейер воспроизводится локально за полчаса: поднимите брокер в Docker из третьего урока, создайте топик orders и его пару orders.dlq, напишите в основной топик смесь валидных JSON и одной строки not-json-at-all. Запустите хендлер со счётчиком из этого урока — увидите в логах три попытки, перенос в DLQ и оживление партиции: события после яд-записи пойдут без рестартов. Затем уберите коммит после отправки в DLQ и получите чистый эксперимент с циклом смерти. Такой стенд нагляднее схем: разница между «с конвейера сняли» и «конвейер встал» — это буквально одна строка CommitMessages.

Правила обращения с яд-сообщениями#

Классифицируйте ошибку до действия: инфраструктура ретраится, данные после порога уезжают в DLQ.
Коммит после успешной отправки в DLQ обязателен: без него цикл смерти продолжится с тем же офсетом.
В DLQ уходит сырое сообщение: значение, ключ, заголовки с причиной и следом ошибки.
Счётчик в памяти обнуляется рестартом и ребалансом: для строгих SLA — retry-топик с числом попыток в заголовке.
Алерт на скорость появления записей в DLQ: всплеск после релиза — инцидент, а не шум.
Паники в хендлере ловятся recover-обёрткой: павший читатель группы приводит партицию к ребалансу, а не к излечению.

Допишите перенос яд-записи

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

const maxFails = 3 if err := json.Unmarshal(m.Value, &order); err != nil { c.fails[string(m.)]++ if c.fails[string(m.Key)] >= maxFails { if err := c..WriteMessages(ctx, kafka.Message{ Key: m.Key, Value: m.Value, }); err != nil { return err } delete(c.fails, string(m.Key)) } return reader.(ctx, m) // перешагиваем яд-запись }

Соберите путь сообщения через хендлер

Расставьте строки: чтение, выход по ошибке, парсинг, валидный путь, счётчик, порог, отправка в DLQ, коммит.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: яд, DLQ, ретраи

Термин

Poison message

нажмите, чтобы перевернуть

Определение

Сообщение, которое падает в хендлере всегда: битый JSON, несовместимая схема. Ретраи не лечат — байты не меняются.

1 / 5

Консьюмер kafka-go доверяет вам всё, что в других очередях спрятано за флагами: классификацию ошибок, число попыток, судьбу необработаемых записей. Цикл смерти распознаётся по повторяющемуся офсету и лечится тремя ветками: ретрай инфраструктуры, порог неудач, DLQ с коммитом и продолжением. Счётчик в памяти — быстрый старт, retry-топик — устойчивый зрелый вариант, а разбор мёртвых писем держится на алерте, осознанном ретеншене и прописанном пути восстановления. Дальше по курсу: урок про коммит офсетов детализирует механику, на которой построен каждый коммит этого урока; урок про ошибки продюсера покажет зеркальную сторону — WriteErrors, с которыми DLQ-Writer может сам оказаться в беде; урок про ребаланс объяснит, почему счётчики в памяти живут ровно до первого переезда партиции; а урок про тесты вырастит этот хендлер в проверяемый конвейер, где DLQ-ветка покрывается сценарием с битой строкой. Начните в песочнице с битой строки в JSON, затем повторите на Docker-брокере: один CommitMessages отделяет живой конвейер от остановленного, и после этого урока вы увидите эту строку в чужом коде с первого взгляда.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Яд-сообщение падает в хендлере всегда: битый JSON или несовместимая схема не исправляются ретраями.
Без обработки яд запускает цикл смерти: читает → падает → перечитывает тот же офсет, партиция стоит.
Ошибки инфраструктуры ретраятся с бэкоффом: сеть и рестарты брокера проходят сами.
Ошибки данных после порога N уезжают в <topic>.dlq сырыми: значение, ключ, причина в заголовке.
Коммит после успешной отправки в DLQ перешагивает яд-запись и оживляет конвейер.
Счётчик в памяти обнуляется рестартом и ребалансом; retry-топик с задержкой устойчивее и дороже.
DLQ читается обычным Reader со своей группой: лог, алерт, ручной разбор и путь восстановления.

Домашнее задание

Соберите локальный стенд из третьего урока: топик orders и orders.dlq. Напишите хендлер из этого урока с unmarshal, счётчиком неудач и переносом в DLQ, затем отправьте в orders смесь валидных JSON и одной битой строки — убедитесь, что после трёх попыток ядовитая запись ушла в orders.dlq, а здоровые события после неё обработались без рестарта процесса. Уберите коммит после отправки в DLQ и зафиксируйте цикл смерти в логах, объяснив, на каком шаге обрывается продвижение офсета. Добавьте алерт-строку на появление записей в DLQ и протокол разборщика: Reader на orders.dlq с группой, лог ключа и заголовка dlq-reason, ручной коммит. Финал: опишите словами, при каком характере ошибок хендлера вы бы перешли от счётчика в памяти к retry-топику с задержкой, и какие заголовки понесло бы сообщение retry-топика.

Вопросы и ответы по уроку

Чем DLQ в Kafka отличается от мёртвых писем в RabbitMQ?

В RabbitMQ dead letter exchange — встроенный механизм брокера: очередь сама перенаправляет отклонённое или умершее по TTL сообщение. В Kafka роли нет: DLQ — это соглашение вашего кода об отдельном топике, перенос делает читатель через Writer, а критерий «проблемно» определяет хендлер. Обратная сторона: DLQ в Kafka — обычный топик с партициями, ключами и ретеншеном, поэтому разборщик может читать параллельно, а порядок внутри партиции сохраняется. Гибкость выше, ответственности больше: забытый коммит после переноса превращает DLQ-механику обратно в цикл смерти.

Сколько попыток ставить в пороге maxFails?

Две-три для ошибок данных достаточно: цель порога — не спасти битые байты, а пережить инфраструктурную совпавшую мгнову — мигнувшую сеть, короткий рестарт зависимости, временно недоступную базу. Больший порог лишь дольше держит офсет на месте и продлевает лаг, потому что каждая попытка яд-сообщения — это пауза для всей партиции. Если хендлер зависит от медленных внешних систем и падает по тайм-аутам, порог не решает проблему: между попытками нужна задержка, а это уже аргумент за retry-топик, где паузу задаёт сам конвейер.

Можно ли просто логировать яд-сообщение и коммитить сразу?

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

Термины урока

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