Шаблон: спецификация Consumer (Kafka, RabbitMQ)

Consumer — компонент, который читает сообщения из брокера (топик Kafka, очередь RabbitMQ), десериализует их, выполняет бизнес-логику и фиксирует прогресс обработки (commit offset / ack). Шаблон фиксирует алгоритм обработки, политику commit/ack, идемпотентность и реакцию на ошибки — для передачи в разработку и QA. Теория брокеров и EDA — 03.05.04 Асинхронное взаимодействие (очереди, Kafka, Events) и 03.05.06 Event-Driven Architecture; контракт сообщения — 03.05.03 Спецификации контрактов (OpenAPI, AsyncAPI, Protobuf) и парный 10.01.06 Шаблон спецификации Producer (Kafka, RabbitMQ).

1. Назначение

Спецификацию Consumer пишет системный аналитик; её читают разработка, QA, платформенная команда и владельцы Producer.

Шаблон закрывает пять задач:

• Подключение и конфигурация — group/queue, prefetch, offset reset, параллелизм; отдельные поля для Kafka и RabbitMQ.
• Политика фиксации прогресса — когда commit/ack относительно побочных эффектов (БД, API, события); единое правило для разработки и приёмки.
• Алгоритм обработки — шаги после валидации, идемпотентность, порядок сообщений.
• Ошибки и DLT/DLQ — retry, poison pill, dead letter; обязательное поведение по offset/ack при отправке в DLT.
• Эксплуатация — метрики, SLI/SLO, критерии приёмки, трассировка к FR/NFR (10.01.15 Шаблон матрицы трассировки, 02.04.02 Критерии приёмки (Acceptance Criteria)).

Граница артефакта: шаблон описывает потребителя сообщений. Не дублирует учебник по Kafka/RabbitMQ — для моделей доставки и паттернов см. 03.05.04 Асинхронное взаимодействие (очереди, Kafka, Events). Исходящие синхронные вызовы (REST) — 10.01.04 Шаблон спецификации REST API; схемы таблиц — 10.01.03 Шаблон спецификации БД сервиса.

Один документ — один брокер и один вход. Kafka и RabbitMQ различаются семантикой ack, DLX и контрактом подписки. Если prod — Kafka, а staging — RabbitMQ «зеркало», оформите отдельную спеку на staging или явный под-scope в «Scope компонентов» с отдельными §3.2–3.3 для каждого брокера. Несколько listener'ов одного брокера на разные топики/очереди — один документ с перечнем ID в §3.0.

2. Когда использовать

Подходит, если:

• сервис подписан на топик Kafka или очередь RabbitMQ и выполняет бизнес-логику по событию;
• нужен человекочитаемый документ с алгоритмом, матрицей ошибок, идемпотентностью и NFR;
• есть или планируется спека Producer — consumer должен ссылаться на тот же контракт.

Не подходит или требует дополнения, если:

• нужен обзор брокеров, offset, ack/nack — 03.05.04 Асинхронное взаимодействие (очереди, Kafka, Events);
• описываете отправителя — 10.01.06 Шаблон спецификации Producer (Kafka, RabbitMQ);
• синхронный REST/gRPC как основной контракт — 10.01.04 Шаблон спецификации REST API;
• платформенные NFR (lag, retention, audit) — 10.01.12 Шаблон нефункциональных требований (NFR), 08.02 Нефункциональные требования (НФТ) — классификация и стандарты;
• TLS/SASL/ACL на канале consume — чеклист в §3.2, детали — 03.05.05 Безопасность API, 08.07 Основы кибербезопасности для системного аналитика;
• PII и чувствительные данные в payload — перечень полей, маскирование в логах, ограничения retention; при регуляторике — 08.06 Юридические и регуляторные требования РФ (PCI-DSS).

Входные условия: утверждённый контракт Producer или AsyncAPI; границы сервиса (03.01.04 Интерфейсы и интеграционные взаимодействия); термины payload (10.01.14 Шаблон глоссария и словаря данных).

3. Структура документа

Скопируйте блок в Confluence или Git. Поля обяз. нужны для утверждения; да* — при выполнении условия в скобках.

3.0. Шапка governance

3.1. Общая информация

Для RabbitMQ «exactly-once» на уровне брокера не гарантируется — используйте at-least-once + идемпотентность или outbox/inbox на стороне приложения.

Exactly-once (Kafka) — минимальный набор полей в спеке:

3.2. Параметры подключения и конфигурация

Kafka (дополнительные поля):

RabbitMQ (дополнительные поля):

Безопасность канала (чеклист, детали — 03.05.05 Безопасность API):

3.3. Политика commit / ack (критичное поле)

Зафиксируйте одно правило для всего consumer. Без этого разработка и QA расходятся в трактовке повторов.

Нормативная формулировка для шаблона (at-least-once):

1. Десериализация и валидация.
2. Claim идемпотентности (§3.7) — запись PROCESSING или эквивалент.
3. Побочные эффекты (БД, API, publish).
4. Статус PROCESSED.
5. Только then — commit offset (Kafka) или ack (RabbitMQ manual).

При падении между шагами 3–4: сообщение будет доставлено повторно; consumer должен безопасно продолжить (идемпотентность §3.7).

Батч (Kafka): commit — последний непрерывный offset по каждой партиции (last contiguous committed offset per partition), а не произвольный набор «успешных из батча». При таймауте max.poll.interval — не commit незавершённых; сообщения перечитаются.

Батч (RabbitMQ): ack по одному сообщению или batch ack только для полностью обработанных delivery tag; nack с requeue=false при отправке в DLX.

DLT / DLQ — обязательное поведение:

3.4. Десериализация и валидация

3.5. Алгоритм обработки

Опишите нумерованный список — он должен совпадать с диаграммой последовательности в §4. Шаги 4–5 и побочные эффекты §3.8 — каркас: состав эффектов определяете по фиче, не копируйте email из примера без необходимости.

Порядок сообщений: если критичен — зафиксируйте ключ партиции (Kafka) или single active consumer (RabbitMQ); иначе — «порядок не требуется».

Rebalance / деплой: при graceful shutdown — завершить in-flight до max.poll.interval; не commit частично обработанных. При rebalance mid-flight — in-flight завершается или откатывается без commit; сообщение будет доставлено повторно.

3.6. Обработка ошибок

Таблица ниже — образец для домена «уведомление по заказу»; строки замените на эффекты из §3.8. Retry: N попыток, политика backoff и jitter — согласовать с платформой (шаблон: exponential backoff, не фиксировать ms as-built).

DLT/DLQ: имя топика/очереди; сохранять исходное тело, headers, timestamp, errorCode.

Retry vs частично выполненный эффект: если внешний сервис не идемпотентен — зафиксируйте барьер: «повтор вызова только до первого успешного ответа; при redelivery при PROCESSING — не вызывать повторно, завершить UPDATE PROCESSED». Статус PROCESSING — барьер от двойного эффекта.

3.7. Идемпотентность и дедупликация

Сценарии:

3.8. Побочные эффекты

Заполните по фиче — таблица ниже иллюстрация для примера §4, не обязательный состав.

3.9. Мониторинг, SLI и алерты

Согласуйте с 10.01.12 Шаблон нефункциональных требований (NFR) и 08.02 Нефункциональные требования (НФТ) — классификация и стандарты: владелец алертов, runbook, целевые SLO.

Трассировка: propagation traceparent / W3C Trace Context и correlationId из headers сообщения — в structured logs и исходящие REST (10.01.12 Шаблон нефункциональных требований (NFR)). Не логировать PII из payload.

Алерты: указать канал (PagerDuty/Slack), владельца команды, ссылку на runbook.

3.10. Версионирование и совместимость

3.11. Отладка (локально / staging)

Только staging / localhost; без prod credentials. Политика — 08.07 Основы кибербезопасности для системного аналитика.

Kafka: отдельная debug consumer group (не prod group) — сообщения остаются для prod consumer.

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic orders.events --group order-service-debug --from-beginning

RabbitMQ (management plugin):

Внимание: rabbitmqadmin get снимает сообщение с очереди (ack). Используйте только dedicated test queue, не общую staging/prod очередь. Для интеграционных тестов предпочтительнее publish тестового сообщения.

rabbitmqadmin get queue=orders.notifications.debug ackmode=ack_requeue_false
# или publish тестового сообщения для локального consumer:
rabbitmqadmin publish exchange=orders.topic routing_key=order.created.v1 \
  payload='{"eventId":"...", "eventType":"OrderCreated", ...}'

3.12. Критерии приёмки (QA)

Формат — 02.04.02 Критерии приёмки (Acceptance Criteria). Сценарии ниже — для примера §4; адаптируйте под §3.8.

• Given валидное OrderCreated → within 5 s эффект выполнен, запись PROCESSED, lag не растёт.
• Given duplicate eventId → эффект один раз, commit/ack без ошибки.
• Given invalid JSON → сообщение в DLT/DLQ, offset/ack зафиксирован, consumer_dlt_total++.
• Given transient 503 внешнего API → retry до N×, затем DLT/FAILED; без двойного эффекта при recovery.
• Given два параллельных consumer → один эффект на eventId.
• Given partial batch (Kafka): в батче два offset, первый OK, второй poison → commit только contiguous до первого; второй → DLT без «дыры» в committed offset.
• Given payload с неизвестным optional полем (forward compatible §3.10) → обработка успешна, поле игнорируется.
• Given replay (новая consumer group / reset offset) при существующих PROCESSED в TTL → дубликаты не вызывают повторных эффектов.
• Given rebalance mid-flight (stop consumer pod) → без commit частичной обработки; после redelivery — идемпотентное завершение.

3.13. Примечания и ограничения (NFR)

4. Пример заполнения

Правило: один документ — Kafka prod. Staging mirror на RabbitMQ — отдельная спека OrderCreatedConsumer-RMQ-staging (ссылка из этой §4 не дублирует §3.2–3.3 RabbitMQ).

Scope: OrderCreatedConsumer в Order Notification Service; Apache Kafka; at-least-once; prod.

Контракт: Producer spec + AsyncAPI orders-events v1.0, channel orders.events, operation onOrderCreated; subject orders.events-value, JSON Schema в Git.

Kafka: топик orders.events (fixed, не pattern), group order-service-notification-group, enable.auto.commit=false, max.poll.records=500, auto.offset.reset=latest, max.partition.fetch.bytes=5 MB.

Политика commit/ack: после шага 5 (PROCESSED); при DLT — commit после publish в orders.events.dlt.

Claim-конфликт: при PROCESSING другим pod — wait 30 s, SELECT каждые 5 s; если PROCESSED — ack; иначе — не ack (redelivery).

Алгоритм (фрагмент): см. §3.5; email POST /email/send с header Idempotency-Key: {eventId}.

Replay: инициатор — platform ops; окно — по заявке; записи PROCESSED в TTL 30 d блокируют повторные эффекты.

Диаграмма последовательности (нотация — 03.02.03 Диаграмма последовательности):

sequenceDiagram
    participant Broker as Kafka
    participant Consumer as OrderCreatedConsumer
    participant DB as PostgreSQL
    participant Email as Email Service

    Broker->>Consumer: Сообщение(ы)
    loop Для каждого сообщения
        Consumer->>Consumer: Десериализация + валидация
        Consumer->>DB: SELECT idempotency_events BY eventId
        alt status = PROCESSED
            Consumer->>Broker: commit offset
        else status = FAILED
            Consumer->>Broker: DLT или skip по runbook
        else не обработано / PROCESSING
            Consumer->>DB: INSERT PROCESSING (claim)
            alt conflict PROCESSING другим worker
                Consumer->>DB: bounded wait + SELECT
                alt стал PROCESSED
                    Consumer->>Broker: commit offset
                else timeout
                    Consumer->>Consumer: не commit — redelivery
                end
            else claim OK
                Consumer->>Email: POST /email/send (Idempotency-Key)
                Email-->>Consumer: 202 Accepted
                Consumer->>DB: UPDATE status=PROCESSED
                Consumer->>Broker: commit offset
            end
        end
    end

5. Связанные артефакты

6. Связи с другими разделами базы

Темы, которые в §3 даны полями, но не раскрыты как теория:

• Гарантии доставки, offset, poison message — 03.05.04 Асинхронное взаимодействие (очереди, Kafka, Events)
• Rebalance, consumer group, EOS-транзакции — 03.05.06 Event-Driven Architecture
• AsyncAPI bindings и lifecycle контракта — 03.05.03 Спецификации контрактов (OpenAPI, AsyncAPI, Protobuf), 03.05.07 Управление версионированием API и жизненным циклом
• Паттерны распределённых систем (outbox, inbox, saga) — 03.06.01 Архитектурные паттерны распределённых систем

7. Источники и дополнительные материалы

Документация брокеров

• Apache Kafka Documentation — Consumer API — https://kafka.apache.org/documentation/#consumerapi
• Apache Kafka — Consumer Configs — https://kafka.apache.org/documentation/#consumerconfigs
• Apache Kafka — Transactions (exactly-once) — https://kafka.apache.org/documentation/#transactions
• Confluent — Kafka Consumer — https://docs.confluent.io/platform/current/clients/consumer.html
• RabbitMQ — Consumer Acknowledgements — https://www.rabbitmq.com/docs/confirms
• RabbitMQ — Dead Letter Exchanges — https://www.rabbitmq.com/docs/dlx

Паттерны и контракты

• Enterprise Integration Patterns — Message Consumer — https://www.enterpriseintegrationpatterns.com/MessageConsumer.html
• Enterprise Integration Patterns — Dead Letter Channel — https://www.enterpriseintegrationpatterns.com/DeadLetterChannel.html
• Idempotent Consumer (Chris Richardson) — https://microservices.io/patterns/communication-style/idempotent-consumer.html
• Confluent — Error Handling and Dead Letter Queues — https://developer.confluent.io/courses/kafka-development/error-handling/
• AsyncAPI Specification — https://www.asyncapi.com/docs
• Confluent Schema Registry — https://docs.confluent.io/platform/current/schema-registry/index.html