Неделя 5: Kafka, сага, Idempotency-Key, AnalyticsService на ClickHouse

С первых четырёх недель сервисы общаются синхронно через gRPC. Для CRUD этого хватает: вызвал, дождался ответа, пошёл дальше. Но долгий процесс (сборка корабля), рассылка уведомлений нескольким получателям и аналитика на потоке событий просят асинхронную модель: отправил событие и не ждёшь. На этой неделе семь задач сразу: добавляем Kafka, AssemblyService, доводим оплату до production-надёжности через Idempotency-Key, замыкаем сагу через OrderCancelled, прокидываем correlation_id через всю цепочку и поднимаем седьмой сервис, AnalyticsService на ClickHouse.

Цели недели

1. Поднять Kafka в KRaft-режиме (без ZooKeeper) одним брокером.
2. Создать AssemblyService с теми же слоями что у предыдущих.
3. Описать контракты событий OrderPaid, OrderCancelled, ShipAssembled с обязательным correlation_id.
4. Реализовать Producer/Consumer в OrderService и AssemblyService через transactional outbox.
5. Обернуть Kafka в platform-библиотеку с middleware для logger, метрик и correlation_id.
6. Перевести PaymentService на Postgres, добавить Idempotency-Key.
7. Замкнуть сагу через OrderCancelled (компенсация при отказе оплаты).
8. Создать AnalyticsService: Kafka in -> ClickHouse, gRPC read API.

Что меняется в системе

Счастливый путь:

OrderService.Pay  --(OrderPaid)-->  Kafka  --+--> AssemblyService.OnOrderPaid
                                              |        |
                                              |        v
                                              |   sleep 10s, "сборка"
                                              |        |
                                              |        v
                                              |   --(ShipAssembled)--> Kafka --+
                                              |                                |
                                              +--> AnalyticsService            +--> OrderService.OnShipAssembled
                                              +--> NotificationService              | status -> COMPLETED
                                                                                     +--> AnalyticsService
                                                                                     +--> NotificationService

Компенсация при отказе оплаты:

OrderService.Pay  --(payment failed)-->
OrderService publishes OrderCancelled  --> Kafka  --+--> InventoryService освобождает резерв
                                                     +--> NotificationService шлёт "Заказ отменён"
                                                     +--> AnalyticsService пишет факт отмены

Это Saga по choreography, где каждый сервис реагирует на событие и публикует следующее, без центрального оркестратора. AnalyticsService подписан на все события и не публикует ничего сам, это терминальный read-model.

Kafka в KRaft-режиме

KRaft, это режим Kafka без ZooKeeper. К версии 3.5 он стал production-ready, а в 4.0 остался единственным: ZooKeeper выпилили совсем. Один процесс совмещает роли controller и broker.

docker-compose

# в deploy/compose/core/docker-compose.yml
services:
  kafka:
    image: apache/kafka:3.7.0
    container_name: kafka
    ports:
      - "${KAFKA_PORT:-9092}:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
      KAFKA_MIN_INSYNC_REPLICAS: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
    networks:
      - microservices_net

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "${KAFKA_UI_PORT:-8090}:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
    networks:
      - microservices_net
    depends_on:
      - kafka

Для dev хватает одного брокера. В продакшене минимум три, плюс replication.factor=3, min.insync.replicas=2.

Создание топиков

docker exec kafka kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create --topic order-paid \
  --partitions 3 \
  --replication-factor 1

Партиций больше единицы даже на dev, чтобы сразу привыкнуть: Kafka держит порядок только внутри одной партиции, не глобально.

Контракты событий

Хранить контракты надо так же серьёзно, как HTTP/gRPC API. Два варианта:

1. JSON Schema, проще, читаемо, удобно для дебага.
2. Protobuf, типизированно, компактно, schema registry для строгой эволюции.

В этом курсе берём JSON: его видно глазами и удобно смотреть в kafka-ui. В реальном продакшене берут Protobuf через Confluent Schema Registry или Apicurio.

contracts/order_service_contracts.md

event: OrderPaid
topic: order-paid
key: order_uuid     # для shard-by-order гарантии порядка
schema:
  type: object
  required: [event_id, occurred_at, order_uuid, user_uuid, amount_cents, payment_method, correlation_id]
  properties:
    event_id:        { type: string, format: uuid }
    occurred_at:     { type: string, format: date-time }
    order_uuid:      { type: string, format: uuid }
    user_uuid:       { type: string, format: uuid }
    amount_cents:    { type: integer, minimum: 0 }
    payment_method:  { type: string, enum: [CARD, SBP, CRYPTO] }
    correlation_id:  { type: string, format: uuid, description: "UUID цепочки запросов от edge до consumer-а" }

event: OrderCancelled
topic: order-cancelled
key: order_uuid
schema:
  type: object
  required: [event_id, occurred_at, order_uuid, user_uuid, reason, correlation_id]
  properties:
    event_id:        { type: string, format: uuid }
    occurred_at:     { type: string, format: date-time }
    order_uuid:      { type: string, format: uuid }
    user_uuid:       { type: string, format: uuid }
    reason:          { type: string, enum: [PAYMENT_FAILED, INVENTORY_UNAVAILABLE, USER_REQUESTED] }
    correlation_id:  { type: string, format: uuid }

Поля которые обязательно есть во всех событиях:

• event_id, для идемпотентности на consumer-стороне,
• occurred_at, бизнес-время события,
• correlation_id, единый ID цепочки запросов между сервисами,
• event_type, имя в формате <aggregate>.<verb> (опционально, если уже есть в topic).

correlation_id, как пробрасывается

Поток выглядит так:

1. Envoy на входе ставит заголовок X-Correlation-Id (если уже есть, переиспользует).
2. OrderService читает заголовок, кладёт в OTel baggage и в request-context.
3. При записи в outbox мы кладём correlation_id прямо в payload.
4. Outbox-воркер шлёт сообщение, не теряя поле.
5. Consumer-ы (Assembly, Notification, Analytics) берут correlation_id из payload в свой ctx, и если публикуют дальше, переносят в исходящие события.
6. В W8 OpenTelemetry: correlation_id шлёт в trace.SetAttributes("correlation_id", id), что позволяет искать trace и логи по одному ID.

В platform/kafka это делается одним middleware: producer берёт correlation_id из ctx и кладёт в payload, consumer обратное.

Producer и Consumer на segmentio/kafka-go

Из двух популярных Go-клиентов (segmentio/kafka-go и IBM/sarama) у kafka-go меньше кода на простой Producer и Consumer, поэтому в курсе берём его. У Sarama больше настроек и режимов, но и порог входа выше:

go get github.com/segmentio/kafka-go

platform/kafka/producer

type Producer struct {
    writer *kafka.Writer
    log    logger.Logger
}

func NewProducer(brokers []string, topic string, log logger.Logger) *Producer {
    w := &kafka.Writer{
        Addr:                   kafka.TCP(brokers...),
        Topic:                  topic,
        Balancer:               &kafka.Hash{}, // shard-by-key
        RequiredAcks:           kafka.RequireAll,
        AllowAutoTopicCreation: false,
        Async:                  false,
    }
    return &Producer{writer: w, log: log}
}

func (p *Producer) Send(ctx context.Context, key string, value any) error {
    payload, err := json.Marshal(value)
    if err != nil { return err }
    return p.writer.WriteMessages(ctx, kafka.Message{
        Key:   []byte(key),
        Value: payload,
        Time:  time.Now(),
    })
}

RequiredAcks=RequireAll плюс min.insync.replicas на брокере, базовая защита от потери сообщений.

platform/kafka/consumer

type Consumer struct {
    reader *kafka.Reader
    log    logger.Logger
}

type Handler func(ctx context.Context, msg kafka.Message) error

func NewConsumer(brokers []string, groupID, topic string, log logger.Logger) *Consumer {
    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers:        brokers,
        GroupID:        groupID,
        Topic:          topic,
        MinBytes:       1,
        MaxBytes:       10e6,
        StartOffset:    kafka.LastOffset,
        CommitInterval: 0, // ручной commit после успешной обработки
    })
    return &Consumer{reader: r, log: log}
}

func (c *Consumer) Run(ctx context.Context, h Handler) error {
    for {
        msg, err := c.reader.FetchMessage(ctx)
        if err != nil {
            if errors.Is(err, context.Canceled) { return nil }
            return err
        }
        if err := h(ctx, msg); err != nil {
            c.log.Error("kafka handler", logger.Err(err))
            continue // в дев-варианте, в проде, retry или DLQ
        }
        if err := c.reader.CommitMessages(ctx, msg); err != nil {
            c.log.Error("kafka commit", logger.Err(err))
        }
    }
}

CommitInterval=0 плюс ручной CommitMessages дают at-least-once гарантию: после успешной обработки коммитим offset.

Идемпотентность на consumer-стороне

Поскольку у нас at-least-once, одно сообщение может прийти дважды. Consumer должен быть идемпотентным:

• по event_id хранить таблицу processed_events (event_id PK, processed_at),
• перед обработкой проверять не было ли,
• бизнес-операции писать в одной транзакции с записью event_id.

Это must-have для production-Kafka, особенно при оплатах и денежных операциях.

Размер сообщения

Kafka спроектирована под поток множества мелких записей, а не под крупные блобы. Поэтому «из коробки» один продюсер не отправит сообщение больше примерно 1 МБ, пока ты явно не поднимешь согласованно несколько лимитов:

• message.max.bytes на брокере и max.message.bytes на топике (исторически около 1 МБ, в свежих версиях 1048588 байт),
• max.request.size на продюсере (дефолт 1048576, ровно 1 МБ),
• replica.fetch.max.bytes на брокере и fetch.max.bytes на консьюмере, иначе крупное сообщение застрянет на репликации или чтении.

На практике сообщения держат меньше 1 МБ, и не из-за самого лимита, а ради производительности:

• Память. Брокер и консьюмер буферизуют сообщение целиком, крупные записи раздувают heap и провоцируют паузы GC.
• Батчинг. Пропускную способность даёт сборка мелких записей в батч и сжатие пачкой, одно жирное сообщение ломает эту экономику.
• Репликация и хвост задержек. Большой батч дольше летит по сети и дольше реплицируется, растёт latency для всех в партиции.

Если полезная нагрузка большая (картинки, отчёты, дампы), берут паттерн claim-check: кладём данные в объектное хранилище (S3 и подобное), а в Kafka шлём только ссылку и метаданные.

Не путай это с MaxBytes: 10e6 в нашем консьюмере выше: там 10 МБ, это размер буфера на один fetch-запрос (сколько байт сообщений притянуть за раз), а не лимит на одно сообщение.