Модуль 4. Консьюмеры Kafka: лаг, ack и пул воркеров

Цель модуля: воспроизвести отставание консьюмера при живом API, прочитать лаг группы и объяснить его скоростью разбора, а не «Kafka тормозит». Починить воркер, вернуть лаг в норму и сформулировать защиту, которая сработает до тикетов от поддержки.

Перед началом: консьюмеры и группы — mq-track kafka m3, пулы горутин — go-concurrency m16, пайплайн и его лаги — pg-clickhouse m6. Инструменты: Docker Compose v2, kafka-consumer-groups.sh, psql, curl, Prometheus, make. Трудоёмкость: ~1.5 ч (разбор 35 мин · практика 50 мин · квиз 10 мин).

🔥 Инцидент: «заказы создаются, но не становятся оплаченными»

11:20, OrderFlow. Поддержка приносит пачку тикетов: заказ создан, 201 Created, но висит в статусе created и не переходит в paid. API отвечает за десятки миллисекунд, 5xx нет, PostgreSQL и реплика здоровы, дашборд молчит: p99, RPS и доля ошибок в норме.

Дежурный смотрит на консьюмера: группа payments показывает LAG около 114 и не двигается. Ночью worker остановили «на минуту», чтобы пересобрать образ, и не вернули. API публиковал события, Kafka их принимала, но обработчик не запущен — статусы застыли, лаг копится.

📖 4.1. Consumer group, партиции и лаг

POST /orders пишет строку в PostgreSQL и публикует событие в топик orders: ключ — user_id, значение — JSON заказа (api/internal/events/events.go). Продюсер включается только при заданном KAFKA_BROKERS; в core-профиле его нет. В full-профиле api шлёт события на kafka:9092, внешний listener — localhost:9094. Топик orders создаётся автоматически с одной партицией по умолчанию. Партиция — единица параллелизма группы: её читает не больше одного участника. В логе партиции порядок сохраняется, между партициями — нет; ключ user_id держит события пользователя вместе.

Группа payments хранит committed offset — позицию подтверждённых сообщений; LAG = LOG-END-OFFSET − CURRENT-OFFSET. Лаг измеряется в сообщениях, а не в секундах: 100 событий при потоке 10/с и 1000/с — разное время. Это один из трёх лагов пайплайна (pg-clickhouse m6); устройство групп и offsets — mq-track kafka m3.

📖 4.2. Ack после обработки: at-least-once и повторная доставка

Worker читает orders группой payments (api/cmd/worker/main.go). Reader создан с CommitInterval: 0: автокоммита нет, offset двигает только явный CommitMessages.

1. FetchMessage получает сообщение, семафор ограничивает число обработчиков.
2. Обработчик «проводит платёж» (time.Sleep(40ms)) и выполняет UPDATE orders SET status='paid' WHERE id=$1 с таймаутом 3 с.
3. Только после успешного UPDATE вызывается CommitMessages; при ошибке — failed++, сообщение не коммитится и будет доставлено снова.

Это at-least-once: «хотя бы один раз», а не «ровно один». Повтор возможен после падения, рестарта, ребаланса или ошибки записи. Защита от дублей — в операции: UPDATE ... WHERE id идемпотентен, поэтому повтор не меняет результат, хотя worker_processed_total вырастет дважды. Две оговорки:

• битое JSON-сообщение не зацикливает worker: worker_failed_total растёт, offset коммитится — poison-сообщение пропускается;
• при параллельных обработчиках коммиты идут не по порядку: если N упало, а N+1 закоммитилось, offset уедет за N и повтора не будет. Строгий at-least-once требует последовательной обработки в партиции — в стенде этого нет.

📖 4.3. Пул воркеров и prefetch: где появляется backpressure

У worker один цикл FetchMessage, а обработка вынесена в горутины, число которых ограничено семафором WORKERS (в compose — 4). Семафор и есть окно in-flight: пока все четыре слота заняты, цикл не берёт следующее сообщение, reader не запрашивает новую пачку, и давление уходит обратно в Kafka — в лаг, а не в очередь в памяти процесса. Пропускная способность пула — WORKERS / время обработки = 4 / 0.04 с = 100 сообщений/с. Если поток выше, лаг растёт линейно, сколько бы ни было партиций; если ниже — стабилен или падает. Prefetch здесь не буфер на тысячи сообщений, как в Java-клиенте: MinBytes: 1, MaxBytes: 10e6, MaxWait: 300ms задают параметры fetch-запроса, а фактическое окно — четыре горутины, которые держат сообщения. Больше WORKERS — выше параллелизм и нагрузка на PostgreSQL; меньше — медленнее разбор, но легче мастеру. Пул горутин и ограничение параллелизма — go-concurrency m16.

📖 4.4. Почему «добавить консьюмеров» не всегда помогает

Параллелизм группы ограничен числом партиций. В orders партиция одна, поэтому:

• второй процесс войдёт в группу, но партицию не получит и будет простаивать; лишний участник добавит только ребаланс;
• четыре горутины внутри процесса параллелят одну партицию, но не дают упорядоченности и усложняют коммиты (см. 4.2);
• масштабирование вширь требует больше партиций, а ключ user_id держит события пользователя вместе: после смены числа партиций «ключ → партиция» меняется.

Второй предел — PostgreSQL: каждый обработчик делает UPDATE в мастер, а пул мастера ограничен (PG_POOL_MAX, по умолчанию 20). При больших WORKERS быстрее упираемся в соединения и блокировки, чем в брокер. Ребаланс тоже стоит денег: пока партиции перераспределяются, группа не читает, и лаг делает горб. «Сколько консьюмеров?» — вопрос про число партиций, скорость прихода и время обработки.

📖 4.5. Диагностика: лаг группы, метрики worker, статусы заказов

Три независимых сигнала, которые складываются в одну картину.

1. Лаг группы. CLI лежит в образе Kafka по абсолютному пути: /opt/kafka/bin не попадает в PATH login-shell.

bash
docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --describe --group payments

LAG по партициям и CONSUMER-ID: у остановленного обработчика потребителя нет, CURRENT-OFFSET стоит, LOG-END-OFFSET растёт. Суммарный лаг check-4 считает отсюда.

2. Статусы заказов. Латентность и 5xx api ничего не скажут: строка в orders записана, событие опубликовано. Считаем по мастеру: SELECT status, count(*) FROM orders GROUP BY status;. Пока worker стоит, новые заказы в created; после догона — в paid.

3. Метрики worker. worker_processed_total и worker_failed_total отдаются на :9102 и скрейпятся Prometheus (job worker); порт не опубликован на хост:

bash
docker compose exec -T worker wget -qO- http://localhost:9102/metrics | grep '^worker_'
curl -s 'http://localhost:9092/api/v1/query?query=worker_processed_total'

up{job="worker"} == 0 — воркер недоступен, rate(worker_processed_total[1m]) — скорость обработки. Панели и алерта на лаг в стенде нет: инцидент виден по тикетам, а не по мониторингу.

📖 4.6. Компромиссы

• At-least-once vs at-most-once. Ручной коммит после обработки не теряет события ценой повторов; коммит до обработки быстрее, но при падении сообщение исчезает. Автокоммит даёт «примерно» и скрывает потери; ручной точнее, но добавляет round-trip к координатору.
• Пул и партиции vs порядок. Больше WORKERS или партиций — быстрее разбор, но выше конкуренция в PostgreSQL, больше перестановок в коммитах и слабее порядок по ключу.
• Retry без backoff vs DLQ. Немедленное перечитывание быстро возвращает успех, но при poison-сообщении крутит горячий цикл; DLQ ограничивает ущерб, но требует ручного разбора.

💻 Практика

Все команды — из каталога polygon/. Время — ориентир для одного человека.

1. Базовая линия (10 мин). Kafka есть только в full-профиле:

bash
make up-full        # первый запуск собирает образ (~3–5 минут)
make seed
docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --describe --group payments

Ожидаемо: LAG = 0. Создайте заказ и убедитесь, что он стал paid:

bash
ID=$(curl -s -X POST localhost:8088/orders -H 'Content-Type: application/json' -d '{"user_id":1,"amount":1000}' | python3 -c 'import sys,json;print(json.load(sys.stdin)["id"])')
sleep 1; curl -s "localhost:8088/orders/$ID?direct=1"

2. Сломать (3 мин). make fail-4 останавливает worker и создаёт 100 заказов. Повторите --describe: LAG около 114, CONSUMER-ID пуст. В psql: SELECT status, count(*) FROM orders GROUP BY status; — новые заказы в created.

3. Красный чек (2 мин). make check-4 опрашивает группу каждые 5 секунд и через 120 секунд печатает CHECK FAIL: лаг группы payments не опустился ниже 50 за 120с — ожидаемо.

4. Диагностика (10 мин). Пока чек ждёт, соберите доказательства: LAG не падает; GET /healthz — 200; p99 и 5xx в Grafana в норме; up{job="worker"} равен 0. API пишет в PostgreSQL и публикует событие, но обработчика нет — отстаёт последнее звено.

5. Починить (2 мин). make fix-4 && make check-4: worker дочитывает 100 сообщений за ~1 секунду (100 × 40 мс / 4), лаг падает до 0, чек зелёный — CHECK OK: сценарий 4: лаг группы payments = 0 (< 50).

6. Упражнение: размер пула (15 мин). WORKERS задан в docker-compose.yml ("4"). Создайте снова 100 заказов и запустите разовый контейнер с другим значением:

bash
make fail-4
docker compose --profile full run --rm -e WORKERS=1 worker

Во втором терминале следите за LAG: при одном обработчике разбор занимает ~4 секунды, при четырёх — ~1 секунду, при восьми — доли секунды, пока не упрётесь в fetch и PostgreSQL. Ctrl-C, затем make fix-4. Два процесса worker на одной партиции не удвоят скорость: партиция достанется одному, второй будет простаивать в группе.

7. Защита (10 мин). Предложите алерт на лаг (в стенде нет экспортёра Kafka — сигнал придётся строить на up{job="worker"} и worker_processed_total, либо добавить экспортёр), SLO «время до paid», идемпотентность и DLQ для poison-сообщений.

🧩 Quiz "Проверь себя"

✅ Итоги модуля

• Поднять full-профиль и увидеть лаг группы payments в --describe.
• Объяснить, чем LAG в сообщениях отличается от «отставания в секундах».
• Показать на стенде, что при остановленном worker растут лаг и число created.
• Пройти цикл fail-4 → check-4 (red) → fix-4 → check-4 (green).
• Объяснить, почему WORKERS ускоряет разбор лишь до пределов партиции и PostgreSQL.
• Сформулировать защиту: алерт на лаг, SLO времени до paid, DLQ, идемпотентность.

🔗 Мостик к Модулю 5

Консьюмер догнал поток, но следующий отказ поджидает на пути чтения. Горячий ключ hot:1 живёт 30 секунд и истекает под нагрузкой; десятки клиентов одновременно видят промах и все идут считать один и тот же отчёт в PostgreSQL. В модуле 5 воспроизведём кэш-шторм, увидим, как cache_compute_total догоняет cache_misses_total, и погасим его singleflight, TTL и джиттером. Навык тот же: измерить, найти звено, починить причину.