А теперь специально всё сломаем: DLQ руками на Go + PostgreSQL

В первых двух постах (тут и тут) мы разобрали, зачем вообще нужна DLQ, почему для неё иногда удобно использовать PostgreSQL и где можно легко выстрелить себе в ногу.

Теперь давайте не будем больше говорить про архитектуру. Лучше один раз всё сломаем руками.

Kafka для этого не понадобится. Возьмём REST API вместо producer, Go-приложение вместо consumer и PostgreSQL для хранения проблемных сообщений.

В итоге хотим получить вот такой сценарий:

отправили сообщение ↓ consumer упал ↓ сообщение попало в DLQ ↓ починили обработчик ↓ сделали replay ↓ сообщение успешно обработалось

Структура проекта будет примерно такой:

. ├── cmd/ │ └── main.go ├── internal/ │ ├── consumer/ │ │ └── consumer.go │ ├── handler/ │ │ └── handler.go │ ├── model/ │ │ └── model.go │ └── repository/ │ └── dlq.go ├── migrations/ │ └── 001_init.sql ├── requests.http ├── docker-compose.yml ├── Makefile └── go.mod

Для демо достаточно обычного docker-compose:

services: postgres: image: postgres:15 environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: dlq ports: - "5432:5432"

Запускаем:

bash docker-compose up -d

Теперь создадим таблицу для сообщений, которые не удалось обработать.

CREATE EXTENSION IF NOT EXISTS pgcrypto; CREATE TABLE dead_letter_queue ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), topic TEXT NOT NULL, partition_key INT NOT NULL, offset BIGINT NOT NULL, message_key BYTEA, message_value JSONB NOT NULL, error_message TEXT NOT NULL, retry_count INT DEFAULT 0, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_dlq_created_at ON dead_letter_queue(created_at); CREATE INDEX idx_dlq_topic ON dead_letter_queue(topic);

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

Для события хватит простой структуры:

type Event struct { EventType string `json:"event_type"` Payload json.RawMessage `json:"payload"` }

А вот для DLQ нам уже нужна дополнительная информация:

type DeadLetterMessage struct { ID uuid.UUID Topic string PartitionKey int Offset int64 MessageKey []byte MessageValue json.RawMessage ErrorMessage string RetryCount int }

Здесь есть важное разделение:

Event — это бизнес-сообщение.
DeadLetterMessage — это уже сообщение плюс технический контекст, который понадобится для расследования.

Сам consumer для демо можно сделать совсем примитивным:

type Consumer struct { dlq *Repository } func (c *Consumer) Process( ctx context.Context, event Event, ) error { if event.EventType == "poison_pill" { return errors.New("cannot process poison pill") } log.Printf("processed event: %s", event.EventType) return nil }

Никакой реальной бизнес-логики здесь нет. Нам нужно только одно: иметь сообщение, которое гарантированно падает.

Поэтому договариваемся, что:

{ "event_type": "poison_pill" }

всегда вызывает ошибку.

Теперь проблему можно воспроизводить сколько угодно раз.

Обычное событие:

POST http://localhost:8080/api/v1/events Content-Type: application/json { "event_type": "user_created", "payload": { "email": "test@example.com" } }

Оно успешно проходит обработку.

Теперь отправляем то самое:

POST http://localhost:8080/api/v1/events Content-Type: application/json { "event_type": "poison_pill", "payload": { "email": "test@example.com" } }

Consumer получает сообщение. Обработка падает. Вместо того чтобы бесконечно пытаться обработать его снова, сохраняем сообщение в PostgreSQL. Теперь оно лежит в DLQ и больше не блокирует основной поток.

Сам repository здесь выглядит довольно скучно:

func (r *Repository) Save( ctx context.Context, msg DeadLetterMessage, ) error { const query = ` INSERT INTO dead_letter_queue ( topic, partition_key, offset, message_key, message_value, error_message, retry_count ) VALUES ($1, $2, $3, $4, $5, $6, $7) ` _, err := r.db.Exec( ctx, query, msg.Topic, msg.PartitionKey, msg.Offset, msg.MessageKey, msg.MessageValue, msg.ErrorMessage, msg.RetryCount, ) return err }

А consumer связывает обработку и DLQ:

func (c *Consumer) Handle( ctx context.Context, msg Message, ) error { event, err := parseEvent(msg.Value) if err != nil { return c.saveToDLQ(ctx, msg, err) } if err := c.Process(ctx, event); err != nil { return c.saveToDLQ(ctx, msg, err) } return nil }

То есть любое сообщение, которое мы не смогли обработать, уходит в отдельный контур.

Но положить сообщение в DLQ — только половина работы

Допустим, через несколько часов мы нашли баг. Исправили код. Теперь нужно вернуть сообщение обратно в обработку.

Добавим:

POST /api/v1/dlq/{id}/replay

И вот здесь есть место, где очень легко сделать ошибку.

Не удаляйте сообщение из DLQ перед обработкой.

Например, такой код выглядит вполне логично:

func (c *Consumer) Replay( ctx context.Context, id uuid.UUID, ) error { msg, err := c.dlq.Get(ctx, id) if err != nil { return err } if err := c.dlq.Delete(ctx, id); err != nil { return err } return c.Process(ctx, msg.Event) }

Но представьте, что Process() упал.

Получаем:

1. достали сообщение 2. удалили из DLQ 3. начали обработку 4. обработка упала 5. сообщения больше нет

Вот вам и потеря данных.

Правильнее сначала обработать сообщение и только после успешного результата удалить его:

func (c *Consumer) Replay( ctx context.Context, id uuid.UUID, ) error { msg, err := c.dlq.Get(ctx, id) if err != nil { return err } if err := c.Process(ctx, msg.Event); err != nil { return err } return c.dlq.Delete(ctx, id) }

Теперь если обработка снова упала — запись никуда не делась. Можно исправить проблему и попробовать ещё раз.

Смотрим, что лежит в DLQ:

GET http://localhost:8080/api/v1/dlq

Берём id сообщения:

POST http://localhost:8080/api/v1/dlq/<ID>/replay

Пока poison_pill считается ошибкой, replay будет падать, а сообщение останется в DLQ.

Теперь исправляем обработчик.

Убираем условие:

if event.EventType == "poison_pill" { return errors.New("cannot process poison pill") }

И запускаем replay ещё раз. На этот раз сообщение успешно обрабатывается. После этого его можно удалить из DLQ.

Получается довольно простой цикл:

ошибка ↓ DLQ ↓ исследование ↓ исправление ↓ replay ↓ успешная обработка ↓ удаление из DLQ

Её всё ещё нет. И это нормально.

Сейчас у нас:

text HTTP → Consumer → PostgreSQL

В реальном сервисе HTTP можно заменить Kafka:

text Kafka → Consumer → PostgreSQL

При этом сама логика обработки проблемных сообщений практически не меняется.

В этом и была идея демо. Не написать «свою Kafka», а отделить получение сообщения от его обработки. Тогда источник сообщений можно заменить, не переписывая весь механизм работы с ошибками.

И да, такой стенд сегодня вполне можно попросить собрать AI-кодером. docker-compose, миграцию, модели и HTTP-ручки он напишет довольно быстро.Но я бы всё равно внимательно посмотрел на три места: миграцию, транзакции и replay.

Потому что создать таблицу dead_letter_queue — действительно пять минут. А вот сделать так, чтобы она через полгода не стала причиной нового факапа, — уже инженерная задача.

А теперь специально всё сломаем: DLQ руками на Go + PostgreSQL

Сначала поднимем PostgreSQL

Теперь немного Go

Ломаем обработчик

Отправляем первое сообщение

Записываем сообщение в DLQ

Проверяем

И где здесь Kafka?