Schema Registry 与数据契约

第27章：Schema Registry 与数据契约

导读：Schema Registry 如何解决数据契约演进？

本章核心问题：Schema Registry 如何解决数据契约演进？

读完本章你将理解：

• 兼容性规则：BACKWARD/FORWARD/FULL
• Avro/Protobuf/JSON Schema 对比
• Schema Registry 架构
• 生产环境最佳实践

Level 1 · 你需要知道的（1-3年经验）

四种兼容性级别

Schema Registry 的核心价值之一是在 schema 注册时执行兼容性检查。兼容性级别决定了"允许哪些类型的 schema 变更"：

BACKWARD（向后兼容，默认）

新 schema 能读取旧 schema 产生的数据。这意味着用新版本 Consumer 读取旧版本 Producer 的消息时不会出错。

允许的变更：

• 删除字段（Consumer 忽略缺失字段）
• 添加带有默认值的可选字段（旧消息缺少该字段，使用默认值）

禁止的变更：

• 添加没有默认值的必填字段（旧消息不包含该字段，无法满足新 schema 的约束）
• 删除有默认值的字段（实际允许，但不建议，会破坏 FORWARD 兼容）
• 更改字段类型（int → string 等）

实践意义：先部署 Consumer（能读新格式也能读旧格式），再部署 Producer（开始写新格式）。滚动升级时不会有窗口期内的解析失败。

FORWARD（向前兼容）

旧 schema 能读取新 schema 产生的数据。旧版本 Consumer 读取新版本 Producer 的消息时不会出错。

允许的变更：

• 添加字段（旧 Consumer 忽略未知字段）
• 删除有默认值的字段（新消息缺少该字段，旧 schema 用默认值）

实践意义：先部署 Producer（开始写新格式），再逐步更新 Consumer。适合 Consumer 更新缓慢的场景。

FULL（完全兼容）

同时满足 BACKWARD 和 FORWARD 兼容。是最严格的级别，实际上限制最大：

允许的变更：

• 添加带有默认值的可选字段
• 删除带有默认值的可选字段

禁止的变更：几乎所有结构性变更。

适合对数据格式稳定性要求极高的场景，如金融数据、合规数据。

NONE（无兼容检查）

不执行任何兼容性检查，允许任意变更。适合开发阶段频繁迭代 schema 时临时使用，不要在生产环境使用。

兼容性级别在 subject 级别配置：

# 设置 subject 的兼容性级别
curl -X PUT http://schema-registry:8081/config/orders-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL"}'

# 查询当前配置
curl http://schema-registry:8081/config/orders-value

# 测试 schema 是否与当前版本兼容（不实际注册）
curl -X POST http://schema-registry:8081/compatibility/subjects/orders-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"Order\",...}"}'

完整的 Avro Producer/Consumer 示例

// Schema 定义（也可以通过 Avro IDL 或 Maven 插件从 .avsc 文件生成类）
String schemaStr = """
    {
      "type": "record",
      "name": "OrderEvent",
      "namespace": "com.example",
      "fields": [
        {"name": "id",       "type": "int"},
        {"name": "amount",   "type": "double"},
        {"name": "user_id",  "type": "int"},
        {"name": "currency", "type": "string", "default": "CNY"},
        {"name": "discount", "type": ["null", "double"], "default": null}
      ]
    }
    """;

// Producer 配置
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
    StringSerializer.class);
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    KafkaAvroSerializer.class);
// Schema Registry 地址
producerProps.put("schema.registry.url", "http://schema-registry:8081");
// 使用 TopicNameStrategy（默认），subject = "orders-value"
producerProps.put("value.subject.name.strategy",
    "io.confluent.kafka.serializers.subject.TopicNameStrategy");
// 如果 schema 已存在，是否自动注册新版本（生产环境建议 false，由 CI/CD 控制）
producerProps.put("auto.register.schemas", false);
producerProps.put("use.latest.version", true);

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(schemaStr);

try (KafkaProducer<String, GenericRecord> producer = 
        new KafkaProducer<>(producerProps)) {
    GenericRecord order = new GenericData.Record(schema);
    order.put("id", 12345);
    order.put("amount", 299.99);
    order.put("user_id", 67890);
    order.put("currency", "CNY");
    order.put("discount", null);
    
    ProducerRecord<String, GenericRecord> record = 
        new ProducerRecord<>("orders", "12345", order);
    
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("Send failed: {}", exception.getMessage());
        } else {
            log.info("Sent to {}-{} @ offset {}", 
                metadata.topic(), metadata.partition(), metadata.offset());
        }
    });
}

// Consumer 配置
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    KafkaAvroDeserializer.class);
consumerProps.put("schema.registry.url", "http://schema-registry:8081");
// 使用具体的 Java 类反序列化（需要通过 Avro Maven Plugin 生成）
consumerProps.put("specific.avro.reader", true);

try (KafkaConsumer<String, OrderEvent> consumer = 
        new KafkaConsumer<>(consumerProps)) {
    consumer.subscribe(List.of("orders"));
    
    while (true) {
        ConsumerRecords<String, OrderEvent> records = 
            consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, OrderEvent> record : records) {
            OrderEvent event = record.value();
            processOrder(event.getId(), event.getAmount(), event.getCurrency());
        }
    }
}

Level 2 · 它是怎么运行的（3-5年经验）

Confluent Schema Registry 的内部架构

Schema Registry 是一个独立的 HTTP 服务，本身不维护任何关系型数据库——它把所有 schema 存储在 Kafka 的一个名为 _schemas 的内部 topic 中：

_schemas topic（紧凑型，replication.factor=3）
  key: "subject" + schema_id（字符串）
  value: schema 定义的 JSON 表示

这个设计利用了 Kafka 的日志压缩（Log Compaction）特性：每个 subject（schema 命名空间）的最新版本 schema 会被永久保留，旧版本也不会被删除（因为老版本的消息可能还在被消费，需要对应版本的 schema 来解析）。

Schema Registry 实例通过消费 _schemas topic 来构建内存中的 schema 缓存，所以它是无状态的——任意一个 Registry 实例崩溃重启，都可以通过重放 topic 重建完整的状态。多个 Schema Registry 实例组成集群时，它们用 Kafka 的主从协议（Leader Election）保证写操作的线性化：只有 Master 节点处理注册请求，写入 _schemas topic，其他节点转发请求或服务读取。

Level 3 · 规范怎么定义的（资深）

没有 Schema Registry 会发生什么

想象这个场景：订单服务的开发团队决定把 amount 字段从 int（单位：分）改为 double（单位：元），同时把 user_id 字段重命名为 customer_id。他们更新了 Producer 代码并部署上线。下游的风控服务、报表服务和通知服务——它们各自独立部署，还在读取旧格式的消息——瞬间崩溃。

这是"隐式契约"的代价：Producer 和 Consumer 之间通过代码约定消息格式，但 Kafka 的消息只是字节，Kafka Broker 不知道也不关心其中的格式。任何一方的格式变化，都可能在毫无警告的情况下破坏另一方。

Schema Registry 解决的核心问题是让消息格式（schema）变成显式的、版本化的、受兼容性约束的共享契约。Producer 在发送消息前必须向 Schema Registry 注册 schema；Consumer 在解析消息时从 Schema Registry 获取对应版本的 schema；任何违反兼容性规则的 schema 变更在注册时就会被拒绝，而不是等到运行时才爆炸。

Wire Format：消息的物理格式

支持 Schema Registry 的 Serializer（如 KafkaAvroSerializer）会按照特定格式序列化消息：

字节 0：     0x00（魔术字节，标识这是 Schema Registry 格式）
字节 1-4：   Schema ID（大端序 int32，4 字节）
字节 5+：    实际序列化的数据（Avro/Protobuf/JSON Schema 格式）

Consumer 端的 Deserializer 读取这 5 个字节的头部，提取 Schema ID，向 Schema Registry 发起 REST 请求（GET /schemas/ids/{id}）获取对应的 schema，然后用该 schema 解析剩余字节。Schema ID 是 Registry 分配的全局唯一整数（从 1 开始递增），不是 schema 的版本号（版本号是 subject 内部的序号）。

这个设计的关键优势：消息不包含 schema 本身，只有一个 4 字节的 ID 引用。相比把完整 schema 嵌入每条消息（如 JSON with schema 的某些实现），节省了大量网络带宽和存储空间，在高吞吐场景下差异尤为显著。

Deserializer 实现了本地缓存——每个 schema ID 只需要向 Registry 发起一次网络请求，之后命中本地缓存，不影响消费延迟。

Subject 命名策略

Schema 在 Registry 中以 subject 为命名空间组织。一个 subject 对应一个 schema 的演化历史（版本 1、版本 2、版本 3……）。不同的命名策略决定了 subject 名称的生成规则：

TopicNameStrategy（默认）

Subject 名 = <topic>-key 或 <topic>-value。

topic: orders
key schema subject:   orders-key
value schema subject: orders-value

优点：简单直观，每个 topic 有一对 subject；每个 topic 的消息格式统一。
缺点：同一个 Avro record type（如 OrderEvent）被多个 topic 使用时，每个 topic 都有独立的 schema 演化历史，容易产生不一致。

RecordNameStrategy

Subject 名 = Avro record 的完全限定类名（namespace.RecordName）。

Avro record: com.example.OrderEvent
subject: com.example.OrderEvent

优点：schema 与 topic 解耦，同一个 record type 无论用在哪个 topic，共享同一个 schema 版本历史；适合多 topic 共用相同 record type 的场景。
缺点：同一个 topic 可以包含不同 record type 的消息（多态），Consumer 需要根据 schema ID 动态选择反序列化策略。

TopicRecordNameStrategy

Subject 名 = <topic>-<namespace.RecordName>，结合了前两者的特点：topic 级别隔离 + record type 级别隔离。适合同一个 topic 承载多种消息类型的场景（如 CDC topic 包含多张表的变更）。

配置命名策略：

// Producer 端配置
props.put("value.subject.name.strategy", 
    "io.confluent.kafka.serializers.subject.RecordNameStrategy");

// Consumer 端配置（通常保持与 Producer 一致）
props.put("value.subject.name.strategy",
    "io.confluent.kafka.serializers.subject.RecordNameStrategy");

Avro Schema 演化示例

安全的 schema 变更

// 版本 1：初始 schema
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example",
  "fields": [
    {"name": "id",     "type": "int"},
    {"name": "amount", "type": "double"},
    {"name": "user_id","type": "int"}
  ]
}

// 版本 2：添加可选字段（安全，满足 BACKWARD 兼容）
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example",
  "fields": [
    {"name": "id",       "type": "int"},
    {"name": "amount",   "type": "double"},
    {"name": "user_id",  "type": "int"},
    // 新增字段必须有默认值，否则旧消息（不包含此字段）无法被新 schema 解析
    {"name": "currency", "type": "string", "default": "CNY"},
    // nullable 字段：union with null，default 为 null
    {"name": "discount", "type": ["null", "double"], "default": null}
  ]
}

破坏性的 schema 变更（会被 BACKWARD 拒绝）

// 版本 3：破坏性变更（以下任一都会被 Schema Registry 拒绝）

// ❌ 删除有值的必填字段（旧消息有 user_id，新 schema 没有，旧 Consumer 无法读取新消息）
// → 违反 FORWARD 兼容

// ❌ 添加没有默认值的字段
{"name": "merchant_id", "type": "int"}  // 没有 default → 旧消息缺此字段 → 报错

// ❌ 更改字段类型（int → string）
{"name": "amount", "type": "string"}    // 类型不兼容 → 旧消息 amount 是 double → 无法解析

// ❌ 重命名字段（从 schema 角度，重命名 = 删除旧字段 + 添加新字段）
{"name": "customer_id", "type": "int"}  // 旧字段 user_id 消失 → BACKWARD 检查失败
// 正确做法：用 aliases 处理重命名
{"name": "customer_id", "type": "int", "aliases": ["user_id"]}  // ✅ 保留别名

数据契约：Schema 作为团队间的 API

Schema Registry 不仅是技术组件，更是团队协作协议的实施机制。它把消息格式从"大家都知道但从未写下来"的隐性约定，变成了有版本控制、有兼容性规则、可以用 CI/CD 工具强制执行的显性契约。

Schema Registry 在 CI/CD 中的实践

在 Producer 服务的构建流水线中，每次 schema 变更都应该运行兼容性检查：

<!-- Maven 插件：在 build 阶段检查 schema 兼容性 -->
<plugin>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-maven-plugin</artifactId>
    <version>7.6.0</version>
    <configuration>
        <schemaRegistryUrls>
            <param>http://schema-registry:8081</param>
        </schemaRegistryUrls>
        <!-- 检查 src/main/avro 目录下的所有 .avsc 文件 -->
        <subjects>
            <orders-value>src/main/avro/OrderEvent.avsc</orders-value>
        </subjects>
    </configuration>
    <executions>
        <execution>
            <id>check-compatibility</id>
            <phase>verify</phase>
            <goals>
                <!-- 检查兼容性，不实际注册 -->
                <goal>test-compatibility</goal>
            </goals>
        </execution>
        <execution>
            <id>register-schemas</id>
            <phase>deploy</phase>
            <goals>
                <!-- 仅在 deploy 阶段（生产部署时）才注册新 schema -->
                <goal>register</goal>
            </goals>
        </execution>
    </executions>
</plugin>

通过这个配置，任何破坏 BACKWARD/FULL 兼容性的 schema 变更，在 mvn verify 阶段就会失败，PR 无法合并，Producer 无法部署。这把兼容性检查从运行时（Consumer 崩溃）提前到构建时（PR review 阶段）。

Protobuf 和 JSON Schema 支持

Schema Registry 不只支持 Avro，还支持：

Protocol Buffers：二进制格式，性能与 Avro 接近，schema 演化规则通过字段号（field number）实现——添加新字段、删除旧字段（只要字段号不复用）都是向后兼容的。优势是语言支持更广泛（C++、Go、Python、Rust 等），且 Protobuf 本身就有良好的 schema 演化文档。

// Protobuf Producer 配置
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    KafkaProtobufSerializer.class);
props.put("schema.registry.url", "http://schema-registry:8081");

JSON Schema：保持 JSON 的可读性，同时通过 JSON Schema 标准（Draft 7）定义数据结构。兼容性检查规则与 Avro 类似。适合不想引入 Avro 依赖但又需要 schema 管理的场景，或者消费方是非 JVM 语言（Node.js、Python）的场景。

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    KafkaJsonSchemaSerializer.class);

三种格式的对比：Avro 序列化最紧凑（schema 不嵌入消息），Protobuf 紧凑且跨语言生态最好，JSON Schema 可读性最好但消息体积最大。在高吞吐 Kafka 场景（每秒数十万消息），Avro 或 Protobuf 的序列化优势很显著——相比 JSON，可以节省 30-70% 的消息大小，直接降低 Kafka 存储和网络成本。

Schema Registry 是 Kafka 生产化的必要组成部分，不是可选的锦上添花。在没有 Schema Registry 的 Kafka 系统中，schema 变更迟早会导致一次生产事故；有了 Schema Registry，schema 变更变成了一个有规则、可预测、可自动化检查的工程流程。

Level 4 · 边界与陷阱（所有人）

以下是与"Schema Registry 与数据契约"相关的常见边界问题和生产陷阱：

陷阱一：忽略默认配置的隐含假设。 许多 Kafka 参数的默认值是为通用场景设计的，在特定业务场景下可能导致性能问题或数据风险。上线前务必逐一审查与本章主题相关的配置项，确认其是否符合业务需求。

陷阱二：缺乏端到端的验证。 理论理解和生产实践之间往往存在差距。建议在预发布环境中构建完整的测试场景，覆盖正常路径和异常路径（如节点宕机、网络分区、磁盘满），验证系统行为是否符合预期。

陷阱三：监控盲区。 本章涉及的机制如果缺乏对应的监控指标和告警规则，问题往往在造成业务影响后才被发现。建议将关键指标纳入监控体系，设置合理的告警阈值。