Kafka-интеграция и wire formats

Проблема: схема на каждое сообщение

В OCF (Object Container File) writer schema хранится один раз в header файла. Все записи в файле разделяют одну схему — overhead минимален. Но в Kafka каждое сообщение независимо: consumer может прочитать любое сообщение без чтения предыдущих.

Вложить полную JSON-схему (сотни байт — килобайты для сложных типов) в каждое сообщение (десятки — сотни байт payload) — неприемлемый overhead. Kafka-сообщение с 100 байтами данных и 2 КБ схемы — это 95% overhead.

Два решения: Single Object Encoding (спецификация Avro) и Confluent Wire Format (de facto стандарт для Kafka). Оба заменяют полную схему компактным идентификатором.

Single Object Encoding

Avro spec определяет формат для хранения одиночных объектов вне OCF контейнера. Вместо полной схемы используется 8-байтовый fingerprint — Rabin hash от Parsing Canonical Form (PCF) схемы.

Parsing Canonical Form (PCF) — нормализованная JSON-форма схемы для вычисления fingerprint. Убирает whitespace, doc, aliases, default — оставляет только структуру. Две эквивалентные схемы с разным форматированием дают одинаковый PCF и одинаковый fingerprint.

Ограничения:

• Fingerprint — 8 байт (64 бита). Коллизии теоретически возможны, хотя для практических объёмов пренебрежимо маловероятны
• Ридер должен иметь локальный реестр схем (fingerprint → schema) — spec не определяет протокол получения схемы по fingerprint
• Не получил широкого adoption в Kafka — Confluent Wire Format доминирует

Confluent Wire Format

Confluent Platform определяет собственный wire format, ставший de facto стандартом для Avro в Kafka. Вместо fingerprint используется 4-байтовый schema ID из Schema Registry — централизованного сервиса хранения схем.

Overhead: 5 байт на сообщение (1 magic + 4 schema ID). Для 100-байтового payload — 5% overhead вместо 2000%. Schema Registry хранит схемы и выдаёт их по ID через HTTP API.

Schema Registry: Produce / Consume Flow

Schema Registry — центральный компонент Kafka-экосистемы для управления схемами. Producer и consumer взаимодействуют с ним при сериализации и десериализации:

Кэширование: Producer кэширует (subject, schema) → ID после первой регистрации. Consumer кэширует ID → schema после первого fetch. В steady state Schema Registry не участвует в пути данных — все вызовы идут из кэша. HTTP overhead только при cold start или появлении нового schema ID.

Subject Naming Strategies

Schema Registry организует схемы по subjects — логическим именам, привязанным к потокам данных. Subject определяет scope для compatibility checks: новая версия схемы проверяется на совместимость с предыдущими версиями того же subject.

TopicNameStrategy (default):

Subject для key: {topic-name}-key
Subject для value: {topic-name}-value

Пример: topic "orders"
 → orders-key (схема ключа)
 → orders-value (схема значения)

RecordNameStrategy:

Subject = полное имя типа (namespace + name)

Пример: com.example.Order → "com.example.Order"
 — один subject для всех topics, где используется этот тип

TopicRecordNameStrategy:

Subject = {topic-name}-{record-name}

Пример: orders + com.example.Order → "orders-com.example.Order"
 — один subject per type per topic

Generic vs Specific Records

Avro предлагает два стиля работы с записями на стороне клиента:

GenericRecord — динамический доступ по имени поля:

GenericRecord user = new GenericData.Record(schema);
user.put("name", "Alice");
String name = (String) user.get("name"); // без type safety

SpecificRecord — code generation из schema, строго типизированные классы:

// Сгенерированный класс из avro-maven-plugin
User user = User.newBuilder()
 .setName("Alice")
 .setId(42L)
 .build();
String name = user.getName(); // compile-time type check

Compatibility Modes — обзор

Schema Registry поддерживает семь режимов совместимости. Каждый определяет, какие изменения допустимы при регистрации новой версии схемы:

Ключевые выводы

1. Single Object Encoding — spec Avro: C3 01 + 8-byte Rabin fingerprint + payload. Не требует внешнего сервиса, но нуждается в локальном реестре схем
2. Confluent Wire Format — de facto стандарт для Kafka: 0x00 + 4-byte schema ID + payload. Overhead — 5 байт на сообщение
3. Schema Registry кэшируется на стороне клиента — в steady state HTTP-вызовов нет, всё из кэша
4. Subject naming определяет scope для compatibility checks: TopicName (default, per-topic), RecordName (per-type), TopicRecordName (per-type-per-topic)
5. GenericRecord — динамический доступ без code generation. SpecificRecord — compile-time type safety с code generation
6. 7 compatibility modes от NONE (без проверок) до FULL_TRANSITIVE (оба направления, все версии)