Итоговый проект: событийный пайплайн

Пайплайн живёт вокруг одного топика orders с тремя партициями. Продюсер пишет событие order.created, ключом служит order_id: hash от ключа выбирает партицию, все события одного заказа ложатся в одну очередь, и порядок внутри заказа гарантирован — порядок между заказами не нужен, и это осознанный размен пятого урока. Читателей два, и они не мешают друг другу: инвентарь слушает в группе inventory, аналитика — в группе analytics, а потребительские группы устроены так, что один и тот же лог читается каждой независимо, со своими офсетами и своим темпом. Внутри группы партиции распределяются между членами, поэтому масштабирование сводится к добавлению инстансов. Таймауты, ребалансы и батчи наследуются из прошлых уроков: здесь они впервые работают в одной программе, и каждое решение проверяется не схемой, а поведением системы при сбое — потерей, повтором или простоем.

Кирпич проекта — урок-источник#

Кирпич	Урок	Роль в проекте
Balancer: &kafka.Hash{} и ключ order_id	05	события заказа — в одну партицию: порядок на заказ
OrderEvent с полем Version	22	версионированный payload переживает эволюцию схемы
GroupID у Reader	07	инвентарь и аналитика читают один лог независимо
FetchMessage + CommitMessages	08	резерв подтверждён — офсет продвинут, не раньше
Счётчик неудач и топик .dlq	19	яд-данные снимаются с конвейера после трёх попыток
ReadLag	21	лаг группы виден вживую, без остановки чтения
контекст с отменой	18	деплой не рвёт обработку на полуслове
уникальный топик на прогон	20	тест не встречает чужие сообщения

Продюсер: событие с версией

type OrderEvent struct {
        Event   string `json:"event"`
        Version int    `json:"version"`
        OrderID string `json:"order_id"`
        SKU     string `json:"sku"`
        Qty     int    `json:"qty"`
}

var writer = kafka.NewWriter(kafka.WriterConfig{
        Brokers:  []string{"localhost:9092"},
        Topic:    "orders",
        Balancer: &kafka.Hash{}, // ключ решает партицию
})

func PublishOrderCreated(ctx context.Context, o Order) error {
        value, err := json.Marshal(OrderEvent{
                Event: "order.created", Version: 2,
                OrderID: o.ID, SKU: o.SKU, Qty: o.Qty,
        })
        if err != nil {
                return fmt.Errorf("order %s: %w", o.ID, err)
        }
        return writer.WriteMessages(ctx, kafka.Message{
                Key:   []byte(o.ID),
                Value: value,
        })
}

Balancer kafka.Hash направляет сообщение по ключу: один заказ — одна партиция, события не перепутаются. Поле Version — якорь эволюции схем: консьюмер читает его первым и понимает, как разбирать остальное, приём из двадцать второго урока. Writer надёжен по умолчанию: NewWriter подставляет RequireAll, ошибка WriteMessages означает недоставку пачки, и сервис возвращает её наверх вместо ложного подтверждения заказа.

Проверка: ключ и партиция

Два события заказа order_id=ord-7351 ушли с интервалом в секунду. Куда они попадут?

Продюсер: контракт события order.created#

Событие — не строка, а контракт. Поле Event называет факт, поле Version сообщает читателю, на какой схеме данных он смотрит: вторая версия упразднила поле Address и добавила PromoCode, и старый консьюмер, встретив Version: 2, обязан либо уметь читать новую форму, либо честно отказать — но не молча взять первые поля, потеряв остальное. Дисциплина из урока про эволюцию схем: номер растёт при несовместимом изменении, совместимые — добавление необязательного поля — проходят без смены версии. Имя события и имя топика пишутся константами рядом с типом payload: опечатка в строковом литерале компиляцией не ловится, а расхождение между продюсером и консьюмером всплывёт в рантайме, когда заказ уже опубликован. Публикация идёт после сохранения заказа в базу: транзакция первична, событие — уведомление о ней. Порядок обратный создал бы заказ, которого нет: консьюмеры зарезервируют товар по данным, которые база отвергла. Ошибку WriteMessages сервис не глотает: заказ сохранён в базе, но событие не ушло — логируем и отдаём в сверочную процедуру, которая перечитает базу и допубликует недостающее.

Консьюмер-инвентарь: обработка до коммита#

Инвентарь подписан группой inventory и делает ровно то, что отработано в восьмом уроке: FetchMessage достаёт запись, reserveStock резервирует SKU на складе, CommitMessages продвигает офсет — строго после успешной обработки. Порядок хранит корректность: коммит до резервирования роняет сервис между шагами — офсет продвинут, товар не зарезервирован, заказ исчез, и Kafka его не вернёт, потому что группа считает работу сделанной. После обработки наоборот: сбой до коммита вернёт запись тому же или другому члену группы, резерв выполнится второй раз, и это уже видно по order_id в логах. Повтор гасится идемпотентностью: резерв по ключу заказа выполняется один раз, вторая попытка помечает дубль, а не списывает товар заново. Ошибки классифицируются по девятнадцатому уроку: сбой сети уходит на ретрай без продвижения офсета, битый payload после трёх неудач — в orders.dlq, о котором ниже, и партиция оживает. Пауза между попытками — короткий sleep с бэкоффом: он отделяет медленного брокера от постоянного брака и не даёт циклу сжечь процессор в холостую.

Цикл инвентаря с DLQ

const maxFails = 3

func runInventory(ctx context.Context, brokers []string) error {
        reader := kafka.NewReader(kafka.ReaderConfig{
                Brokers: brokers,
                GroupID: "inventory",
                Topic:   "orders",
                MinBytes: 1,
                MaxBytes: 10e6,
                MaxWait:  time.Second,
        })
        defer reader.Close()

        dlq := kafka.NewWriter(kafka.WriterConfig{Brokers: brokers})
        defer dlq.Close()

        fails := 0
        for {
                m, err := reader.FetchMessage(ctx)
                if err != nil {
                        return ctx.Err() // читателя закрыли: чистый выход
                }
                if err := reserveStock(m); err != nil {
                        fails++
                        if fails < maxFails {
                                continue // офсет на месте: запись вернётся заново
                        }
                        _, err = dlq.WriteMessages(ctx, kafka.Message{
                                Topic: "orders.dlq", // Topic в сообщении: у Writer его нет
                                Key:   m.Key,
                                Value: m.Value, // сырьё, без пересборки
                                Headers: []kafka.Header{
                                        {Key: "dlq-reason", Value: []byte("reserve x3")},
                                },
                        })
                        if err != nil {
                                return err // DLQ недоступна: коммитить нельзя
                        }
                }
                fails = 0
                if err := reader.CommitMessages(ctx, m); err != nil {
                        return err
                }
        }
}

Цикл собирает три ветки девятнадцатого урока: успех — коммит; временный сбой — офсет на месте, запись придёт заново; третий провал — сырьё уходит в orders.dlq с причиной в заголовке, и лишь затем CommitMessages перешагивает яд. Writer для DLQ создан без поля Topic: назначение записано в каждом сообщении, и один Writer обслуживает разные топики — иначе kafka-go вернёт ошибку конфликта конфигурации.

Живое место: коммит перед обработкой

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

Проверка: цена преждевременного коммита

Резерв товара поставлен до reserveStock, CommitMessages вызван перед ним, сервис упал между двумя шагами. Что с заказом?

Консьюмер-аналитика: тот же лог, своя группа#

Аналитика читает те же события, но живёт в группе analytics: свои офсеты, свой темп, своя раскладка партиций. Инвентарь встанет на паузу — аналитика не заметит; аналитика отстанет на час — инвентарь продолжит резервировать. Fanout через группы стоит нуля копирований: один физический лог обслуживает сколько угодно независимых потребителей, и новый подписчик подключается без единой правки в продюсере. Настройки чтения у групп разные сознательно: инвентарю важен короткий MaxWait — резерв не должен ждать сбора пачки; аналитике важнее крупный MinBytes — статистика глотает десятки тысяч событий в секунду и берёт их батчами из десятого урока. Обработчик аналитики проще некуда: разобрать payload по версии, обновить счётчики, закоммитить. Ей не нужен строгий realtime: отставание на минуты не искажает суточные срезы, поэтому крупный MinBytes и батчи уместнее, чем погоня за каждым сообщением. Но и здесь ошибка данных после порога уходит в собственный DLQ: витрина, построенная на потоке с ядом, врёт во всех отчётах сразу.

Что вернёт ReadLag у группы?

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

// Reader с GroupID: "inventory", топик orders.
// Написали 40 000 событий, обработали 39 200.
lag, err := reader.ReadLag(ctx)
fmt.Println(lag, err)

Наблюдаемость и остановка: лаг и контекст#

Конвейер, который нельзя наблюдать, нельзя запускать в проде. Лаг — его пульс: ReadLag у обоих читателей раз в десять секунд, число уезжает в метрику, рост без спада — алерт, потому что каждая единица лага — заказ, который ещё не обработан. Второй сигнал — лог самого цикла: ключ, ветка ошибки, длительность обработки; без него разбор инцидента превращается в угадывание по косвенным признакам. Остановка идёт через контекст из восемнадцатого урока: сигнал завершения отменяет корневой ctx, FetchMessage возвращается с ошибкой контекста, цикл выходит, defer закрывает Reader и DLQ-Writer, а необработанный офсет остаётся не закоммиченным — после рестарта группа начнёт с него и ничего не потеряет. Деплой превращён в процедуру: сначала перестаём принимать заказы по HTTP, затем отменяем консьюмеров, затем ждём закрытия Writer, у которого могут остаться незавершённые записи. Таймауты вокруг внешних вызовов — из того же урока: без них одна медленная партиция удержит весь пайплайн. Метрика лага без порогов бесполезна: настраиваются алерт на затяжной рост и абсолютный предел, выше которого конвейер заведомо не успевает за потоком заказов.

Тест: уникальный топик на прогон#

Конвейер получает интеграционный тест из двадцатого урока, и его главный приём — изоляция: имя топика собирает fmt.Sprintf с UnixNano, двух прогонов с одинаковым названием не бывает даже при параллельном запуске пакетов. Тест сам создаёт топик через CreateTopics с NumPartitions: 3 — автосоздание брокера отдало бы конфигурацию дефолтам, а порядок здесь проверяется явно: три события одного заказа уходят продюсером, группа inventory читает их, и резерв вызывается строго в порядке отправки. Второй сценарий — ветка DLQ: в поток подмешивается payload, который reserveStock всегда отвергает, и тест ждёт, что после трёх неудач сырьё окажется в orders.dlq с заголовком-причиной, а следующее событие обработается без рестартов. t.Cleanup регистрирует DeleteTopics сразу после создания: отложенные вызовы срабатывают при любом исходе, включая t.Fatalf, и брокер остаётся чистым. Гонки между продюсером и читателем снимаются FetchMessage с таймаутом контекста — тем же приёмом, что и в боевом цикле, поэтому тестовая обвязка совпадает с продовой почти строка в строку.

Про песочницу: конвейер вживую

Интерактивы исполняются в песочнице курса на программах без сети, а живой прогон воспроизводится локально: брокер в Docker из третьего урока, CreateTopics создаёт orders и orders.dlq с тремя партициями, продюсер шлёт десяток заказов — два Reader с разными GroupID показывают, как инвентарь резервирует, аналитика копит статистику, и оба скроллят один лог со своей скоростью. Прервите читателя по ctrl-c в середине обработки и запустите снова: группа продолжит с последнего закоммиченного офсета — нагляднейшая демонстрация контракта «обработка до коммита».

Чек-лист курса#

Ключ решает порядок: один заказ — одна партиция, kafka.Hash на продюсере.
Событие — контракт: имя константой, Version в payload, совместимые изменения без смены версии.
Обработка до коммита: не обработано — офсет на месте; повторы гасит идемпотентность по order_id.
Группы дают fanout: новый потребитель подключается без правок продюсера.
Ошибки классифицируются: инфраструктура ретраится, данные после порога уходят в .dlq с причиной в заголовке.
Лаг — метрика номер один: ReadLag на каждом читателе, алерт на затяжной рост.
Остановка через контекст: отмена, закрытие Reader и Writer, рестарт продолжит с не закоммиченного офсета.
Тесты изолируются: уникальный топик на прогон, CreateTopics и DeleteTopics в t.Cleanup.

Соберите продюсера проекта

Допишите балансер, который направляет события по ключу, и имя события этого проекта.

writer := kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: "orders", Balancer: &kafka., }) event := OrderEvent{ Event: "order.", Version: 2, OrderID: o.ID, SKU: o.SKU, Qty: o.Qty, }

Соберите цикл инвентаря

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

Строки

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

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

Карточки: узлы пайплайна

Термин

order.created

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

Определение

Событие о создании заказа: имя-константа, JSON payload с полем Version, ключ order_id в kafka.Message.

1 / 6

Проект собран, и это уже не учебная программа: один топик, две независимые группы, DLQ, наблюдаемость и аккуратная остановка — каркас, который в проде дорастает кластером и реестром схем, не меняя формы. Что дальше. Задачи со сроком и статусом — напоминание, отложенная отправка, повтор по расписанию — комфортнее в asynq: там приоритеты, ретраи и планировщик встроены, и Kafka для них избыточна. Кэш витрины и горячие остатки — территория Redis: событийный пайплайн пишет в базу, Redis закрывает чтение, и связка «событие — кэш» разобрана в отдельном курсе. Если вы прошли уроки по порядку — у вас на руках работающий стенд и карта типовых граблей; откройте свой проект, создайте один топик и начните с продюсера, а остальное достраивайте по чек-листу выше. Каркас переносится как отдельный пакет: продюсер, две группы и DLQ-логика не завязаны на функцию main, и переезд в сервисную структуру сводится к разрезанию на модули.

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

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

Вывод

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

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

Пайплайн: заказ → событие order.created → топик orders на три партиции → две группы консюмеров.
Ключ order_id и kafka.Hash дают порядок на заказ без глобальной упорядоченности.
Payload версионирован: поле Version говорит читателю, по какой схеме разбирать остальное.
Инвентарь: FetchMessage, резерв, CommitMessages после обработки; идемпотентность гасит повторы.
Аналитика живёт в своей группе: тот же лог, свои офсеты, ноль копирований.
Ошибки данных после трёх неудач уходят в orders.dlq с причиной в заголовке, затем коммит.
ReadLag на каждом читателе — метрика и алерт: рост лага виден раньше жалоб.
Остановка через контекст: отмена, закрытие Reader и Writer, рестарт продолжит с не закоммиченного офсета.

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

Соберите конвейер на Docker-брокере из третьего урока: CreateTopics создаст orders с NumPartitions: 3 и orders.dlq, продюсер опубликует серию заказов, группы inventory и analytics прочитают лог независимо. Вбросьте в поток запись, которую reserveStock всегда отвергает, и убедитесь по логам, что после трёх неудач она оказалась в DLQ с заголовком-причиной, а партиция продолжила работу. Отдельно проверьте остановку: прервите консьюмера посреди обработки и запустите заново — обработка продолжится с не закоммиченного офсета. Финал — прогон теста из этого урока против того же стенда. Усилить: повесьте ReadLag на тикер раз в десять секунд и объясните, почему Lag() у вашего читателя возвращает минус один.

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

Чем итоговый пайплайн отличается от продовой версии?

Масштабом и краями, а не архитектурой: в проде топик с репликами и RequireAll из пятнадцатого урока, конфиги читателей выверены, DLQ под мониторингом и ретеншеном из семнадцатого, а счётчик неудач переживает рестарт — он в базе, а не в памяти процесса. Схема событий живёт в реестре схем, а не в комментарии рядом с типом, а мониторинг из двадцать первого урока — лаг, рост DLQ, длительность обработки — уезжает на дашборд. Каркас же тот же: ключ, порядок, обработка до коммита, DLQ, лаг, контекст.

Почему два консьюмера, а не один, который делает резерв и статистику сразу?

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

Сохранится ли порядок событий одного заказа при росте числа инстансов?

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

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

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