Схемы событий и Schema Registry

Kafka хранит байты записи, а приложения согласуют их структуру и смысл. Если поле размера было числом 12500, а стало строкой "12500", корректный JSON может оказаться непригодным для прежнего consumer. Схема и правила совместимости помогают обнаружить такое изменение до обработки данных.

Схема описывает форму, контракт добавляет смысл

Схема задаёт поля, их типы и допустимую структуру. Контракт дополнительно объясняет значение: размер в байтах, время в UTC, event_id стабилен при повторе. Смена единицы размера с байтов на килобайты при сохранении типа int формально может пройти проверку схемы и всё равно сломать отчёт.

Для самостоятельного примера записи зададим контракт: event_id — непустая строка, entity_id — положительное целое, size_bytes — неотрицательное целое, encoding — строка raw или gzip. В Python bool является подклассом int, но True не является корректным размером записи. Проверять можно точный тип type(value) is int.

def valid_size(event):
    value = event.get("size_bytes")
    return type(value) is int and value >= 0
 
print(valid_size({"size_bytes": 12500}))  # True
print(valid_size({"size_bytes": "12500"}))  # False
print(valid_size({"size_bytes": True}))  # False

JSON-парсер проверяет синтаксис. Функция проверяет поле нашего контракта. Avro, Protobuf и JSON Schema предоставляют разные способы описания и сериализации структуры; правила эволюции у них различаются. Собственный маленький валидатор ниже не заменяет эти форматы.

Registry хранит версии и проверяет совместимость

Schema Registry является отдельным сервисом для хранения и управления схемами. Confluent Schema Registry часто используют вместе с Kafka; он не встроен в Apache Kafka broker. Брокер по-прежнему хранит key и value как байты. Подключение Registry происходит через согласованные сериализаторы и десериализаторы клиентов.

В Registry схему регистрируют в subject, логическом наборе версий для конкретного назначения данных. Например, стратегия имён по топику может использовать records-value для value. Стратегии имён существуют разные, поэтому нельзя автоматически считать один топик ровно одним subject.

В одном распространённом wire format сериализатор помещает ID схемы рядом с байтами данных. Consumer использует этот ID для выбора схемы. Точный формат зависит от согласованного клиента и сериализации.

Клиенты узнают, с какой writer schema записаны данные, и получают нужную схему из Registry или кеша. Проверка новой версии при регистрации помогает отвергнуть несовместимое изменение до публикации. Но произвольный producer, отправляющий сырой JSON без такого протокола, может обойти этот механизм. Контракт включает и формат байтов, и политику клиентов.

Backward и forward смотрят в разные стороны

Backward compatibility означает, что новый читатель способен прочитать данные предыдущей схемы. Forward compatibility означает, что прежний читатель способен прочитать данные новой схемы. Full требует обе стороны. Транзитивная проверка сравнивает со всеми предшествующими версиями subject, а не только последней.

Рассмотрим отдельный пример Avro record Record. Первая версия содержит event_id и size_bytes. Во второй добавляется поле encoding типа string с default raw. Когда новый reader читает старые данные, он берёт default для отсутствующего в writer schema поля. Старый reader может проигнорировать добавленное поле при чтении новых данных. Этот пример относится именно к правилам Avro, а не автоматически к любому JSON Schema.

Default в Avro помогает разрешению reader/writer schemas. Он не означает, что producer может всегда пропустить поле при записи новой схемой: это зависит от сериализатора и подготовки объекта. В Python-задаче мы сами явно добавляем encoding по явно заданному правилу версии 1; это отдельное преобразование.

Добавить обязательное поле без пригодного default опасно для старой истории: новый reader не знает, откуда взять значение. Изменить числовое поле на строку также нельзя считать безопасной заменой только из-за возможности вызвать int. Перед выпуском проверяют и совместимость формата, и смысл значения, и код обработки.

Обновление читателей связано с replay

Producer и consumer обновляются в разное время. Во время перехода в журнале одновременно остаются события нескольких схем. Новый consumer должен читать ту историю, которая ему требуется, а старый — пережить новые записи до обновления, если это предусмотрено планом.

При backward-переходе обычно сначала обновляют читателей, способных читать старые и новые данные, затем producer. Если replay должен охватить несколько лет и много версий, проверки только последней версии недостаточно. Нужны транзитивная политика или явные тесты каждой сохраняемой writer schema.

Внезапное отсутствие Registry может повлиять на регистрацию новой схемы и получение неизвестной схемы, даже если брокеры исправны. Кеш известных схем смягчает часть обращений; конкретное поведение зависит от клиента. Поэтому наблюдать нужно и доступность сервиса схем, и ошибки сериализации, и версии записей.

Наконец, одинаковый event_id с разным телом является конфликтом бизнес-контракта. Проверка типов не обнаружит, что один идентификатор прислан с разными значениями size_bytes, например 12500 и 7000. Такой конфликт следует выделить отдельно от точного повтора, чтобы защита приёмника не скрывала порчу данных.

Практика

В Python реализуй три функции: проверку описанной записи, добавление encoding к старой версии без изменения входного словаря и поиск конфликтующих тел одинакового event_id. В условиях задано, какие поля сравниваются; timestamps и headers не входят в этот маленький пример.

Тесты проверяют функции на явных словарях. Они не обращаются к Registry, не сериализуют Avro и не устанавливают совместимость настоящей схемы. Для последнего потребуется выбранный формат, сервис Registry и тестовые сообщения клиентов.