Работа с Kafka

В Go-сервисах Ensi Kafka используется для обмена событиями между сервисами. Типичные сценарии такие: мастер-система опубликовала оффер или изменила бренд, маркетинговый сервис отправил команду пересчитать скидки, ваш сервис должен подхватить изменение и обновить локальный кэш или индекс. Вы либо публикуете сообщение в топик, либо подписываетесь на топик и обрабатываете входящий поток, либо делаете и то и другое.

Kafka подключают для сценариев, которые использует конкретный сервис. Далее описан адаптер app/adapters/kafka из Go-шаблона и текущих Go-сервисов. Команды запускаются из корня сервиса через elc.

Правила именования топиков и контуров описаны в Kafka Design Guide. Ниже — как писать и читать сообщения в сервисе на Goravel.

Код разделён по слоям так же, как остальная инфраструктура сервиса.

• Общий клиент, registry хэндлеров, supervisor консьюмеров и продюсер лежат в app/adapters/kafka.
• Обработчики конкретных фич живут рядом с модулем, в app/modules/{module}/kafka.
• Список «какой ключ топика обслуживает какой хэндлер» собирается в bootstrap/kafka.go.
• Параметры брокера, карта ключей на полные имена топиков и список слушаемых ключей по умолчанию задаются в config/kafka.go и переменных окружения.

Топики: короткий ключ вместо полного имени

В прикладном коде почти никогда не пишут настоящее имя топика вроде local.catalog.fact.offers.1. Такое имя зависит от контура и домена и легко разъезжается между окружениями. Вместо этого в config/kafka.go, в секции topics, для каждого топика заводят короткий ключ — offers, brands, examples и так далее — и сопоставляют ему полное имя. Полное имя обычно собирают из KAFKA_CONTOUR (local, dev, prod и т.п.) и соглашения об именовании платформы.

И продюсер, и консьюмер принимают именно этот короткий ключ. При отправке или подписке адаптер сам резолвит его в реальное имя топика на брокере. Поэтому новый топик сначала добавляют в kafka.topics, и только потом начинают ссылаться на ключ из кода, из kafka.listen и из констант модулей. Если ключа нет в карте, запись или чтение закончатся ошибкой конфигурации, а не «тихим» уходом в несуществующий топик.

Список listen в том же конфиге отвечает на вопрос, какие ключи слушаются по умолчанию: их подхватывает встроенный runner при старте web-процесса и команда kafka:consume, если её вызвали без аргументов. Отдельно задают offset_reset — для новых consumer group обычно оставляют beginning, чтобы группа не «пропустила» историю, пока вы явно не решите иное.

Для локальной разработки можно явно включить KAFKA_AUTO_CREATE_TOPICS=true: адаптер сможет создать топик при подключении consumer или публикации. В конфигурации шаблона значение по умолчанию — false. В общих контурах stage/prod топики обычно уже заведены инфраструктурой, и этот флаг лучше держать выключенным, чтобы случайно не плодить лишние топики с дефолтными настройками партиций.

Подключение к брокеру задаётся через переменные окружения.

• KAFKA_BROKER_LIST — список брокеров через запятую. При пустом списке встроенный runner не запускается, а явно вызванный consumer или producer возвращает ошибку конфигурации.
• KAFKA_CONSUMER_GROUP_ID задаёт группу консьюмера; если переменная не задана, используется значение APP_NAME из конфигурации.
• KAFKA_CONSUMER_ENABLED включает встроенный консьюмер вместе с HTTP-приложением.
• TLS и SASL настраиваются через KAFKA_SECURITY_PROTOCOL и KAFKA_SASL_*. Текущий connectionOpts использует SASL/PLAIN; чтение KAFKA_SASL_MECHANISMS из env само по себе не добавляет поддержку других механизмов.
• Сжатие исходящих сообщений настраивают через KAFKA_PRODUCER_COMPRESSION_CODEC (часто snappy).

Как отправить сообщение

Публикация делается через фасад, из хэндлера, artisan-команды или джобы очереди — из любого места, где есть контекст запроса или фоновой задачи:

err := facades.Kafka().Produce(ctx, "topic-key", payload)

После контекста передаётся ключ топика из kafka.topics, затем тело сообщения. Как сериализуется тело, зависит от типа значения. Если передать структуру или map, адаптер закодирует её в JSON. Если передать уже готовые []byte или строку, байты уйдут в брокер как есть.

Ключ сообщения Kafka (partition key) задают согласно контракту топика. Стабильный ключ помогает направлять события одной сущности в одну партицию; сам по себе он не гарантирует общий порядок публикаций из нескольких процессов. Например:

err := facades.Kafka().Produce(
    ctx,
    "offers",
    payload,
    appkafka.WithKey(strconv.Itoa(offerID)),
)

Текущий producer использует ProduceSync и возвращает ошибку отправки вызывающему коду. Успешный вызов подтверждает публикацию в рамках настроек producer, но не обработку consumer-ом. Ошибку обязательной публикации нужно передавать дальше; SQL-транзакция не может откатить уже отправленное сообщение.

Как читать сообщения

Входящие сообщения обрабатывает хэндлер — обычная структура с методом Handle(ctx context.Context, msg appkafka.Message) error. Хэндлеры конкретной предметной области кладут в app/modules/{module}/kafka/, рядом с actions и остальным кодом модуля, а не в adapters: adapters дают транспорт, модуль знает, что делать с полезной нагрузкой.

Минимальный вид хэндлера такой:

type OfferHandler struct{}

func (h *OfferHandler) Handle(ctx context.Context, msg appkafka.Message) error {
    var payload OfferChanged
    if err := msg.JSON(&payload); err != nil {
        return err
    }
    // дальше — бизнес-логика, чаще всего через action или sync модели
    return nil
}

Метод msg.JSON разбирает тело как JSON в переданную структуру. Если нужен сырой вид, байты доступны в msg.Value, ключ сообщения — в msg.Key. Поля Topic, TopicKey, номер партиции и offset полезны для логов и диагностики: по ним видно, какое именно сообщение взяли в работу.

С точки зрения консьюмера сообщение считается обработанным только когда Handle вернул nil. Если метод вернул ошибку, offset за это сообщение не сдвигают. После перезапуска или переназначения партиции сообщение может прийти снова с последнего подтверждённого offset. Встроенного цикла повторного запуска упавшего consumer в supervisor нет. Поэтому обработку стоит писать идемпотентной: повторный create не должен плодить дубли, повторный update не должен портить уже согласованное состояние, повторный delete при отсутствующей записи обычно считается успешным исходом.

Ошибка разбора JSON — тоже ошибка хэндлера, если вы её возвращаете из Handle. «Проглотить» битое тело молча обычно плохая идея: контракт либо чинят на стороне продюсера, либо явно решают политику для poison message, а не оставляют тихий пропуск.

Регистрация хэндлера

Мало написать Handle: хэндлер нужно зарегистрировать в registry. Иначе при старте консьюмер откажется слушать топик — для ключа из listen не окажется обработчика. Регистрация сосредоточена в bootstrap/kafka.go: туда передают ключ топика и экземпляр хэндлера.

registry.Register(brandskafka.TopicKey, brandskafka.NewHandler())
registry.Register(offerskafka.TopicKey, offerskafka.NewHandler())

События изменения моделей (model events)

Многие топики Ensi несут не произвольный JSON «на усмотрение автора», а стандартизованное событие изменения модели. В теле есть тип события (create, update или delete), список полей, которые изменились (dirty), и атрибуты сущности (attributes). Для create/update обычно передаётся полный снимок; для delete контракт может содержать снимок перед удалением или сокращённый набор полей. Такой конверт уже описан в адаптере (ModelEvent, DecodeModelEvent), и модульные хэндлеры обычно декодируют его в типизированный payload:

{
  "event": "update",
  "dirty": ["name", "updated_at"],
  "attributes": { }
}

Типичный ход обработки такой. Хэндлер декодирует envelope, открывает транзакцию ORM и ищет локальную запись по внешнему идентификатору мастер-системы (brand_id, offer_id и т.п.). Для create/update при отсутствии строки создаёт новую, заполняет поля из attributes и сохраняет; для delete удаляет, если запись ещё есть. Это схема полной синхронизации, используемая, например, для локальных реплик в Catalog Cache. Обработчик статусов может обновлять только существующую запись и игнорировать create/delete, как в OMS; определяйте политику по контракту события. Часто после успешного сохранения помечают связанные сущности к переиндексации или другой отложенной работе. На update многие хэндлеры смотрят на пересечение dirty с «интересными» полями: если пришло только то, что локально не используется, сохранение и побочные эффекты можно пропустить.

Повторяющийся каркас find / fill / save / delete в сервисах часто выносят в общий sync-action, чтобы каждый топик описывал только маппинг своей сущности. Как устроены модели и транзакции — в работе с базой данных. Как писать component-тесты на такие хэндлеры (вызов Handle с синтетическим Message, без живого брокера) — в автотестах.

Контракт payload и dirty

До реализации handler-а сверяйте producer schema и реальные сообщения, а не только локальную ORM-модель. Для каждой операции определите:

Часть контракта	Что установить
Identity	ID сущности producer-а, локальный lookup key и Kafka key; они могут различаться
Attributes	Полный snapshot, patch или projection для удаления
Поля	JSON-имена, required/nullable, enum, единицы, числовой диапазон и время
Dirty	Какие изменения интересуют consumer и что означают null/пустой список
Повтор	Create существующей записи, update отсутствующей, no-op, повтор delete
Эффекты	Связанные SQL-записи, маркировка индекса, HTTP/Kafka/storage и их отказ

Для snapshot-события dirty служит условием обработки, а не списком единственных полей для присваивания: Fill переносит все поля снимка, которыми владеет consumer. Payload может описывать больше полей, чем сохраняется локально. Поля другого владельца не нужно затирать из-за отсутствия их в контракте.

Dirty []string принимает JSON null как nil slice. Типизированный DecodeModelEvent[T] декодирует JSON, но не проверяет required, допустимый event или бизнес-смысл ID. Эти проверки выполняет sync-action/handler по контракту. Нулевой ID после decoding не доказывает, что producer прислал валидный идентификатор.

Полная реплика и обновление собственной сущности

Общий sync helper есть в Catalog Cache и OMS в app/modules/common/actions/sync_model.go. Его наличие в другом сервисе проверяйте отдельно.

Handler	Поведение
Catalog Cache, brand	Lookup brand_id = payload.ID, create при отсутствии, перенос snapshot, маркировка офферов при сохранении
OMS, delivery-order	Только update статуса; lookup локальной доставки по payload.DeliveryID, игнорирование неизвестной доставки и create/delete

payload.ID в событии delivery-order — ID сущности Logistic; это не локальный Delivery.ID. Применение полного sync без предварительных условий могло бы создать или удалить сущность, которой владеет OMS.

Семантика SyncModelAction

Следующая таблица описывает текущий общий helper полной синхронизации. Политика конкретного handler-а может дополнительно ограничить вход до его вызова.

Event/состояние	Действие helper-а
Пустой/неподдерживаемый event	Ошибка до выполнения deletion predicate и записи
Delete, строка отсутствует	Успех без записи
Delete, строка существует	Delete callback
Create, строка существует	Fill, затем Save только при изменении
Create, строка отсутствует	Create, Fill, Save
Update с интересным dirty, строка отсутствует	Create, Fill, Save
Update без интересного dirty	Пропуск до lookup
Пустой список interesting fields	Любой update интересен
Dirty null/пустой при непустом interesting list	Update пропускается
Существующая строка, mapped значения прежние	Нет Save и связанных с ним эффектов

При наличии IsToDelete predicate выполняется для поддерживаемого non-delete event до dirty filtering. Так Catalog Cache может удалить опубликованное значение свойства по deleted_at даже без обычного delete event. Не используйте этот callback как универсальную обработку любого невалидного payload.

HasInterestingChanges не сравнивает данные модели: он проверяет пересечение имён полей. За no-op отвечают Fill и его changed. ChangeSet, SetField и SetPointerField помогают учитывать false, 0 и nullable по значению, а не по адресу указателя. Общий helper не содержит проверки версии события: повтор одинакового snapshot и запоздавший старый snapshot — разные случаи. Если нужно отклонять устаревшие события, требуется предусмотренный контрактом механизм версии/порядка.

Транзакционный handler

У Catalog Cache brands точка входа имеет вид:

func (h *Handler) Handle(ctx context.Context, msg appkafka.Message) error {
    return commonactions.HandleModelEvent(ctx, msg, "brand", h.sync)
}

appkafka — goravel/app/adapters/kafka, commonactions — goravel/app/modules/common/actions. Helper декодирует ModelEvent[BrandPayload] и вызывает h.sync(ctx, tx, event.Meta(), event.Attributes) внутри ORM-транзакции.

Все persistence callbacks используют полученный tx: FindModel ищет по контрактному ключу, SaveModel выбирает create/save, Delete удаляет нужную строку. В brands после сохранения вызывается marker.WithQuery(tx).MarkBrand(...), чтобы изменение модели и SQL-маркировка офферов откатывались вместе. Возврат ошибки из callback должен доходить до Handle.

Командные топики имеют собственные сообщения и правила. Для команды пересчёта или запуска задачи не нужно искусственно конструировать model-event только ради переиспользования SyncModelAction.

Публикация model events

Сначала найдите существующие publisher и observer/caller. В CMS nameplates публикация подключена через ORM observers, в OMS ряд действий публикует события явно. Повторная отправка из action поверх observer создаст лишнее событие.

Доменный publisher собирает attributes из нужного состояния модели, затем передаёт типизированный envelope общему producer. Схема вызова для CMS:

event := appkafka.ModelEvent[nameplatekafka.NameplatePayload]{
    Event:      appkafka.ModelEventUpdate,
    Dirty:      dirty,
    Attributes: attributes,
}
return facades.Kafka().Produce(
    ctx,
    nameplatekafka.NameplateTopicKey,
    event,
    appkafka.WithKey(strconv.FormatUint(attributes.ID, 10)),
)

Это фрагмент publisher-а, возвращающего error: attributes уже собран как NameplatePayload, dirty вычислен из реальных изменений; nameplatekafka — пакет модуля nameplates. В действующем коде переиспользуйте PublishNameplate, который уже выполняет это отображение.

Create может требовать ID/timestamps после записи, delete — снимок до удаления, update — вычисленные поля и корректный dirty. API DTO нельзя автоматически считать payload события: набор полей, видимость и сериализация могут различаться. Зафиксируйте порядок dirty, отсутствие дублей и поведение при no-op по действующему контракту.

Wire-сериализация

Текущий encodeBody использует encoding/json/v2 с FormatNilSliceAsNull(true) и FormatNilMapAsNull(true). Поэтому nil dirty сериализуется как null, а явно пустой slice — как []. Строка/[]byte отправляется без повторного JSON-кодирования; plain nil body отвергается. Теги, enum, числа, даты и собственный MarshalJSON нужно проверять на фактических Value bytes.

appkafka.Fake вызывает тот же encoder и сохраняет независимые копии сообщений в порядке публикации. Проверка Go struct до отправки не подтверждает совпадение JSON wire-контракта, а fake не проверяет наличие topic mapping и доступность брокера.

Initial-event и контекст

Producer берёт initial-event из входящего context и добавляет X-Initial-Event. Если context содержит событие, переданные вручную headers с точно таким именем заменяются одним контекстным значением. Сопоставление имени на публикации регистрозависимо; не добавляйте тот же header другим регистром.

Consumer распознаёт имя header без учёта регистра и восстанавливает initial-event в context до Handle. Невалидное непустое значение возвращает ошибку до прикладной обработки. При прямом вызове handler-а в тесте этот транспортный этап не выполняется: Message.Headers сам по себе не наполняет context.

Передавайте полученный context дальше в ORM, HTTP и исходящий Kafka. context.Background() внутри action разрывает цепочку инициирующего события. Для теста propagation создавайте известный initial-event и проверяйте точное значение исходящих headers; для проверки декодирования входящего header нужен тест адаптера.

SQL и доставка

Способ публикации	Что может произойти при отказе
Внутри SQL-транзакции	Ошибка публикации откатит SQL при возврате ошибки, но отправленные ранее сообщения останутся; commit тоже может отказать после успешной публикации
После commit	Данные уже сохранены, ошибка Kafka требует предусмотренного восстановления
Через существующий outbox	Намерение пишется тем же tx; relay, повтор и дедупликация проверяются отдельно

Сохраняйте выбранную архитектуру при портировании. Успешный ProduceSync не означает, что consumer применил событие. Ошибка или timeout тоже не доказывают, что сообщение не попало в брокер. Уникальный Kafka key не является механизмом дедупликации consumer-а.

Для нескольких сообщений учитывайте уже опубликованный префикс. Например, OMS delivery-order может успешно отправить deliveries, затем получить ошибку отправки orders и откатить SQL. Повтор входящего события снова отправит deliveries. Проверяйте и SQL rollback, и сохранённый внешний эффект, затем успешный replay; не называйте эту границу exactly-once.

Проверка сценария

Компонентный тест вызывает реальный action/observer либо Handle с сериализованным envelope и реальной PostgreSQL-БД. Для исходящих событий проверяйте topic key, key bytes, Value, headers, число и порядок сообщений. Для rejected/no-op веток явно проверяйте отсутствие публикации и лишних записей.

В ComponentSuite исходящие сообщения записывает s.Kafka, а общий TearDownTest отклоняет непроверенные публикации. Используйте s.AssertKafkaMessages(topicKeys...) для точного порядка топиков и получения сообщений, затем проверяйте key, сериализованный payload и headers. Вызов без аргументов проверяет отсутствие сообщений. Чтение s.Kafka.Messages() не отмечает фазу проверенной: после собственного assertion-helper нужен s.ConfirmKafkaMessages(), который сам ничего не проверяет.

Для SQL-фикстур используйте s.CreateQuietly(...), если их model events не входят в сценарий. Между фазами сначала проверьте события, затем вызовите s.ResetKafka(), сбрасывающий и сообщения, и отметку проверки. Ручной suite должен зарегистрировать t.Cleanup(s.TearDownTest) после setup; переопределённый teardown должен вызывать базовый. Эти правила действуют и для consumer-теста: входящий Handle может сам публиковать новые события.

Минимальная матрица consumer-а: create/update/delete по его политике, relevant/irrelevant dirty, missing record, повтор, true→false, заполненное→null, malformed JSON, unsupported event, ошибка SQL/эффекта и replay. Она не заменяет проверку topic map, registry, listen и включения runner.

Подробнее о fake, lifecycle suite и injection отказа — в автотестах. Для реализации используются implement-kafka-model-event и publish-kafka-model-event из .agents/skills.

Как запускают консьюмеры

Есть два рабочих способа слушать топики; выбор зависит от окружения.

Первый — вместе с web-приложением. Если Kafka runner зарегистрирован в bootstrap/app.go, выставлены KAFKA_CONSUMER_ENABLED=true и непустой KAFKA_BROKER_LIST, при старте сервиса запускается встроенный consumer. Список kafka.listen должен быть непустым, с зарегистрированными хэндлерами. Встроенный runner создаёт один consumer на этот список. Так удобно локально держать «всё в одном процессе»: подняли сервис — и HTTP, и консьюмер уже работают.

Второй способ — artisan-команда kafka:consume. Её запускают рядом с приложением или вместо встроенного runner:

elc ./artisan kafka:consume
elc ./artisan kafka:consume examples
elc ./artisan kafka:consume offers brands --threads=3

Без аргументов команда слушает те же ключи, что указаны в kafka.listen. Аргументы позволяют сузить набор до конкретных ключей из kafka.topics — это нужно при запуске консюмеров в stage/prod окружении, чтобы в разных k8s подах слушались разные топики. Флаг --threads задаёт число consumer-ов на весь переданный список топиков в одном процессе. Например, offers brands --threads=3 создаёт три consumer-а, каждый подписан на оба топика. Значение меньше 1 заменяется на 1. Три пода с --threads=2 создают суммарно шесть consumer-ов; реальная загрузка зависит от числа партиций и их распределения в группе. Отдельная команда kafka:consume не зависит от флага встроенного runner KAFKA_CONSUMER_ENABLED.

Идентификатор группы берётся из KAFKA_CONSUMER_GROUP_ID, а если он не задан — из APP_NAME. Consumer-ы с одним group id делят партиции между собой. Сбои между выполнением handler и подтверждением offset всё равно могут привести к повторной доставке. Процессы с разными group id читают одни и те же топики независимо: каждый ведёт свой offset.

При старте в stdout обычно печатается список ключей и соответствующих им полных имён топиков — по нему сразу видно, не ошиблись ли в конфиге. По каждому обработанному сообщению пишется прогресс в духе RUNNING / DONE / FAIL вместе с ключом топика и offset.

Остановить консьюмер можно Ctrl+C локально или SIGTERM в Kubernetes. Чтобы быстро проверить связку end-to-end на учебной фиче, в одном терминале запускают kafka:consume, в другом — команду публикации в тестовый топик: консьюмер должен взять сообщение и завершить Handle без ошибки.

Ошибки и повторная обработка

Текущий consumer отключает auto commit. После успешного Handle он явно вызывает CommitRecords. При ошибке handler offset не подтверждается; при ошибке commit обработчик уже мог применить изменения, но подтверждение осталось неизвестным.

В обоих случаях consumer приостанавливает проблемную партицию, закрывает свой клиент и завершает Run с ошибкой. Supervisor не создаёт replacement consumer и не выполняет автоматический retry. Остальные consumer-ы могут продолжать работу, а общий Run ждёт их завершения, поэтому один сбой не обязательно приводит к немедленному выходу всего процесса. Контролируйте ошибки и прогресс consumer group; после устранения причины требуется восстановить работу consumer-а.

Повторная доставка должна быть безопасной и после успешного выполнения с неподтверждённым offset. Уже зафиксированные изменения БД и внешние вызовы не откатываются при ошибке Kafka. Политику poison messages, повторов и восстановления определяют отдельно для конкретного сценария.

Остановка без обрыва текущего сообщения

При получении сигнала остановки (SIGINT или SIGTERM) консьюмер не бросает текущее сообщение на середине. Он даёт уже начатому Handle завершиться, но новые сообщения из брокера больше не берёт. Так проще сохранить согласованность: либо обработка дошла до конца и offset можно сдвинуть, либо процесс убили уже после graceful-фазы — и тогда сообщение обработается снова после старта нового пода.

Это напрямую влияет на настройки Kubernetes. По умолчанию после SIGTERM кластер через примерно тридцать секунд принудительно убивает контейнер. Если ваш хэндлер работает дольше, нужно увеличить terminationGracePeriodSeconds для консьюмеров и воркеров с запасом на это время. Иначе при деплое новой версии сервиса сообщение могут оборвать на полпути, а при следующем старте оно приедет повторно. На время graceful shutdown старый под остаётся в состоянии Terminating и ждёт завершения текущего Handle — это ожидаемое поведение, а не «зависший» деплой.

Исходные примеры

• Catalog Cache: brands handler и sync helper — полная реплика и SQL-маркировка.
• OMS: delivery-order consumer — обновление собственной доставки и несколько публикаций.
• CMS: nameplates publisher и observer.
• CMS: producer и fake — encoder, keys и initial-event.

Пример узкой проверки из корня OMS Go:

elc go test ./app/modules/orders/tests/kafka --env=.env.testing -count=1
elc go test ./app/adapters/kafka --env=.env.testing -count=1
elc go build -o build/app