Сериализация: JSON в сообщениях

Вернёмся к анатомии сообщения из второго урока с новой стороны. kafka.Message несёт несколько полей, и для сериализации значимы три: Key — маршрут по партициям, Value — полезная нагрузка, Headers — пары ключ-значение для метаданных. Поле Value объявлено как []byte, и это не упрощение kafka-go, а честная позиция брокера: Kafka не знает про ваши домены и не обязана знать. Связка «kafka json golang» устроена поэтому тривиально: json.Marshal(order) возвращает байты, они кладутся в Value, консьюмер вызывает json.Unmarshal(m.Value, &event) — и событие снова структура. Сериализационного слоя в kafka-go нет, и это плюс: библиотека не тянет ни protobuf-рантайм, ни avro-схемы, а решение о формате остаётся за вами и меняется без вопросов к клиенту. Заметим и цену: JSON избыточен — имена полей повторяются в каждом сообщении, числа без схемы гадают типы, а деньги во float64 живут с погрешностью. Для сервисных потоков это допустимо, для больших объёмов — нет; альтернативы разберём ниже, а пока зафиксируем правило: то, что вы положили в Value, интерпретирует только ваш же код, и чем раньше оба конца согласуют форму, тем дешевле каждое следующее изменение.

Тег json — это спецификация формы, и у него есть поведение, которое лучше знать до инцидента. Без тега encoding/json использует имя поля как есть: OrderID превратится в «OrderID» с заглавной буквы, а консьюмер на другом языке ждёт order_id — поэтому теги пишут на обеих сторонах, даже если кажется, что и так совпадёт. omitempty пропускает пустые значения при маршаллинге, но ноль — легитимное значение для суммы, скидки или счётчика: если ноль нужно отличать от отсутствия, берите указатели, а не флаг. Неэкспортированные поля не сериализуются вовсе: маленькая буква в имени поля — и поле исчезло из сообщения без предупреждений. При декодировании неизвестные поля игнорируются — это фундамент обратной совместимости: продюсер добавляет поля, и старый консьюмер не падает. Но игнорирование работает в обе стороны: опечатка в теге не вызовет ошибки ни у кого, события просто потеряют поле. Самый дешёвый тест контракта — unit-тест, который маршалит и тут же декодирует структуру и сверяет все поля: он же документирует событие лучше любого комментария.

Идея общего пакета проста, а экономия огромна: структуры событий живут в одном пакете — events, contracts, как ни назовите, — и оба сервиса импортируют его. Один процесс маршалит, второй декодирует, форма существует в единственном экземпляре, и правка проходит одним коммитом через оба сервиса, если они в одном репозитории. Внутри монорепозитория это самая дешёвая схема из возможных: компилятор сам проверяет использование полей, а rename-рефакторинг переименовывает и продюсер, и консьюмер за один заход. С отдельными репозиториями честнее сразу признать: контракт — внешний интерфейс, его изменение — релиз с миграцией, а не правка. Вот где встаёт поле Headers из kafka.Message: тип события и версия схемы едут в заголовках, Value декодируется по ним, и два формата спокойно сосуществуют в одном топике. Полноценный schema registry kafka-go не предоставляет — при потребности подключают внешний реестр или protobuf; но для большинства сервисных потоков общий пакет закрывает задачу целиком, а заголовки дают место для развилки на время перехода.

JSON остаётся дефолтом, потому что читается человеком, дебажится консольными утилитами и не тянет зависимостей, — но у него есть пределы, и лучше узнать о них до внедрения. Первый — размер: имена полей повторяются в каждом сообщении, и на потоках в десятки тысяч событий в секунду вы платите трафиком и диском за повторение order_id миллион раз. Второй — скорость: рефлексия encoding/json не бесплатна, а Writer копит батчи по байтовому лимиту из десятого урока — чем меньше сообщение, тем больше событий уезжает в одном produce-запросе. Третий — типы: у JSON нет десятичных дробей, время живёт строкой или числом, и каждая такая вещь требует конвенции. Альтернативы старые как сам Kafka: protobuf даёт компактность и строгие типы за счёт генерации кода из схемы, avro — эволюцию схем с реестром, msgpack — компромисс по размеру без отдельной схемы. kafka-go честно равнодушен: Value останется []byte при любом выборе, менять придётся только места Marshal и Unmarshal — ещё один аргумент за общий пакет из прошлого раздела. Начинайте с JSON, переходите тогда, когда цифры, а не мода, скажут, что формат стал узким местом.

Худший сценарий сериализации происходит не на продюсере, а на консьюмере: одно сообщение, которое не декодируется, — и группа стоит. Механика проста: ReadMessage возвращает событие, Unmarshal падает, код логирует ошибку и делает continue — но оффсет ещё не закоммичен, и после рестарта группа начнёт с того же места. Топик зацикливается на битом сообщении, лаг растёт, дежурный смотрит в лог с тысячей одинаковых строк. Отсюда правило: ветка ошибки декодирования — это решение о судьбе события, а не строчка лога. Варианты — пропустить с ручным коммитом оффсета, отправить в карантинный топик или ждать фикса с паузой; развёрнутый разбор с DLQ-паттерном — в девятнадцатом уроке. json.Valid позволяет отсечь грубую порчу до Unmarshal, а залогированные топик, партиция и оффсет — найти сообщение консольной утилитой и понять, что в нём лежит. В песочнице это не воспроизвести: нужен локальный брокер из Docker-стенда третьего урока — там и накатайте сценарий с битым Value.