未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

Schema Registry - 数据模式管理与演进

假设一个数据契约失控场景：上游系统未经协调修改用户数据 Schema，下游 20 个消费者系统因不兼容变更全部报错，团队难以定位变更，也不敢轻易回滚。这里的系统数量与事故情节仅用于说明风险，不代表已记录的真实经历。Schema Registry 正是为解决这类"数据契约"失控问题而生。

Schema Registry 的核心价值

生产者与消费者解耦

没有 Schema Registry 时，生产者和消费者需要共享同一个 Schema 定义文件，用带外（out-of-band）方式同步。任何一方改动都可能导致另一方崩溃。

Schema Registry 引入了中央化的 Schema 存储：生产者注册 Schema 并获得一个数字 ID，消费者按 ID 从 Registry 拉取 Schema 自行解析。双方只依赖 Registry，不再直接耦合。

消息编码格式：Magic Byte + Schema ID

本文使用 Confluent KafkaAvroSerializer 序列化非 null Value 时，字节流采用以下 5 字节头部；这不是任意 Schema Registry 或 Kafka 消息的通用格式：

字节 0    : Magic Byte，固定值 0x00，标识 Confluent 序列化格式
字节 1-4  : Schema ID（4 字节大端整数）
字节 5+   : Avro 二进制编码的真实 Payload

消费者收到消息后，先读 Magic Byte 确认格式，再用 Schema ID 从 Registry 获取 Schema，最后反序列化 Payload。这意味着消息体自带"解析说明书"的索引，极大降低了解析复杂度。

Confluent Schema Registry vs Apicurio Registry

维度	Confluent Schema Registry	Apicurio Registry
许可	Confluent Schema Registry 使用 Confluent Community License，不能将 Community 等同于 Apache 2.0；具体发行包及商业功能须按对应许可区分	Apache 2.0；具体发行版本及依赖仍需分别确认
背后公司	Confluent	Red Hat
支持格式	Avro、JSON Schema、Protobuf	Avro、JSON Schema、Protobuf、OpenAPI 等
存储后端	Kafka Topic（_schemas）	内存、PostgreSQL、Kafka 等
REST API 兼容性	原版 API	兼容 Confluent API（需开启）
适用场景	Confluent 生态优先	多注册表格式、Red Hat 生态

本文示例以 Confluent Schema Registry 为基准，API 风格通用。

---

配置与启动

schema-registry.properties 核心参数

# 监听地址，8081 是默认端口
listeners=http://0.0.0.0:8081

# 关联的 Kafka 集群（Schema 持久化存储在 _schemas Topic）
kafkastore.bootstrap.servers=kafka-01:9092,kafka-02:9092,kafka-03:9092

# _schemas Topic 的副本数，生产环境建议与 Broker 数量对齐
kafkastore.topic.replication.factor=3

# 全局默认兼容性级别
schema.compatibility.level=BACKWARD

# 是否参与主节点选举（多实例部署时所有节点设为 true）
master.eligibility=true

# Kafka 存储初始化超时时间，单位毫秒
kafkastore.init.timeout.ms=60000

启动与健康检查

# 启动 Schema Registry（确保 Kafka 已运行）
bin/schema-registry-start config/schema-registry.properties &

# 检查服务是否就绪，返回空数组 [] 表示正常
curl http://localhost:8081/subjects

# 查看全局兼容性配置
curl http://localhost:8081/config

---

Avro Schema 编写

基础结构：Record、Field、Type

Avro Schema 是 JSON 文档，核心元素如下：

{
  "type": "record",
  "name": "User",
  "namespace": "com.example.schema",
  "doc": "用户基本信息",
  "fields": [
    {
      "name": "id",
      "type": "string",
      "doc": "用户唯一标识，UUID 格式"
    },
    {
      "name": "username",
      "type": "string",
      "doc": "用户名，3-20 个字符"
    },
    {
      "name": "age",
      "type": ["null", "int"],
      "default": null,
      "doc": "年龄；本例使用 null 默认值，并将 null 放在 union 首位"
    },
    {
      "name": "created_at",
      "type": {"type": "long", "logicalType": "timestamp-millis"},
      "doc": "创建时间，毫秒时间戳"
    }
  ]
}

常用 Avro 类型速查

分类	类型	示例
基本类型	null / boolean / int / long / float / double / bytes / string	"type": "string"
逻辑类型	date / time-millis / timestamp-millis / decimal	"logicalType": "timestamp-millis"
复杂类型-数组	array	{"type": "array", "items": "string"}
复杂类型-Map	map	{"type": "map", "values": "string"}
复杂类型-枚举	enum	{"type": "enum", "name": "Status", "symbols": ["ACTIVE", "DELETED"]}
复杂类型-嵌套	record	嵌套另一个 record 定义
联合类型	union	["null", "string"]；为兼容要求默认值匹配 union 首个分支的实现，使用 null 默认值时将 null 放在首位

带嵌套结构的完整示例

{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.events",
  "doc": "订单事件，用于 order-events Topic",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "user_id",  "type": "string"},
    {
      "name": "status",
      "type": {
        "type": "enum",
        "name": "OrderStatus",
        "symbols": ["CREATED", "PAID", "SHIPPED", "COMPLETED", "CANCELLED"]
      },
      "default": "CREATED"
    },
    {
      "name": "items",
      "type": {
        "type": "array",
        "items": {
          "type": "record",
          "name": "OrderItem",
          "fields": [
            {"name": "sku",      "type": "string"},
            {"name": "quantity", "type": "int"},
            {"name": "price",    "type": {"type": "bytes", "logicalType": "decimal",
                                          "precision": 10, "scale": 2}}
          ]
        }
      },
      "default": []
    },
    {
      "name": "metadata",
      "type": {"type": "map", "values": "string"},
      "default": {},
      "doc": "扩展属性，key-value 对"
    },
    {
      "name": "event_time",
      "type": {"type": "long", "logicalType": "timestamp-millis"}
    }
  ]
}

---

Schema 兼容性：四种模式对比

兼容性规则定义了新版本 Schema 与旧版本之间的读写关系，是 Schema 演进的核心概念。

四种兼容性模式

模式	含义	新 Consumer 读旧数据	旧 Consumer 读新数据	典型场景
BACKWARD（默认）	新 Schema 能读旧数据	可以	不保证	先升级消费者再升级生产者
FORWARD	旧 Schema 能读新数据	不保证	可以	先升级生产者再升级消费者
FULL	双向兼容（BACKWARD + FORWARD）	可以	可以	滚动升级，最严格
NONE	不做兼容性检查	不保证	不保证	开发/测试环境快速迭代

Schema 修改规则清单

BACKWARD 兼容下允许的操作：

• 添加有 default 值的新字段
• 删除旧字段：BACKWARD 下，新 reader 会忽略旧 writer 中已删除的字段，不要求该旧字段带 default；若同时要求 FORWARD，则需考虑旧 reader 的默认值
• 将 string 类型扩展为 ["null", "string"] union

BACKWARD 兼容下禁止的操作：

• 添加没有 default 的新字段（新 reader 读取旧数据时，无法为旧数据中不存在的新字段补值）
• 删除没有 default 的旧字段并不违反 BACKWARD；但旧 reader 读取新数据时可能因缺少默认值而违反 FORWARD
• 不符合 Avro reader/writer 类型解析规则的类型变更；旧 writer 的 int 可以提升为新 reader 的 long，因此 int 改为 long 可满足 BACKWARD，但不保证 FORWARD
• 直接修改字段名称且未提供可用的 reader alias 或默认值；Avro 按名称及 reader alias 解析字段，改名需按具体读写方向检查
• 删除旧 writer 可能写出的枚举符号，且新 reader 没有可用的枚举默认值；仅调整 symbols 顺序并不必然破坏兼容性，解析会结合 writer schema 按符号名称匹配

Schema 演进三步示例：

// v1：初始 Schema
{"fields": [
  {"name": "id",       "type": "string"},
  {"name": "username", "type": "string"}
]}

// v2：BACKWARD 兼容，新增可选字段（有 default）
{"fields": [
  {"name": "id",       "type": "string"},
  {"name": "username", "type": "string"},
  {"name": "email",    "type": ["null", "string"], "default": null},
  {"name": "version",  "type": "int", "default": 2}
]}

// v3：FULL 兼容，删除有 default 的字段（email 被弃用）
{"fields": [
  {"name": "id",       "type": "string"},
  {"name": "username", "type": "string"},
  {"name": "version",  "type": "int", "default": 2}
]}

---

Java 客户端集成

Maven 依赖

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.6.0</version>
</dependency>

生产者配置（KafkaAvroSerializer）

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-01:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "io.confluent.kafka.serializers.KafkaAvroSerializer");

// Registry 地址，Serializer 会自动注册/查询 Schema
props.put("schema.registry.url", "http://schema-registry:8081");

// 幂等性，生产环境建议开启
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");

KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);

// 加载 Schema 并构造记录
Schema schema = new Schema.Parser().parse(
        new File("src/main/avro/OrderEvent.avsc"));

GenericRecord record = new GenericData.Record(schema);
record.put("order_id", UUID.randomUUID().toString());
record.put("user_id",  "user-123");
record.put("status",   new GenericData.EnumSymbol(
        schema.getField("status").schema(), "CREATED"));
record.put("event_time", Instant.now().toEpochMilli());
record.put("items", Collections.emptyList());
record.put("metadata", Collections.emptyMap());

producer.send(new ProducerRecord<>("order-events", record.get("order_id").toString(), record));
producer.flush();

消费者配置（KafkaAvroDeserializer）

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-01:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        "io.confluent.kafka.serializers.KafkaAvroDeserializer");

props.put("schema.registry.url", "http://schema-registry:8081");

// false = 返回 GenericRecord；true = 返回生成的 SpecificRecord 类
props.put("specific.avro.reader", false);

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("order-events"));

while (true) {
    ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, GenericRecord> r : records) {
        GenericRecord value = r.value();
        System.out.printf("order_id=%s, status=%s%n",
                value.get("order_id"), value.get("status"));
    }
    consumer.commitSync();
}

---

REST API 完整操作

注册 Schema

# REST 示例使用独立的 rest-order-events-value Subject，不与 Java 示例的 order-events Topic 混用
# 向 Subject 注册一个新版本的 Schema
# Subject 命名遵循策略，默认 TopicNameStrategy：<topic-name>-value
curl -X POST http://localhost:8081/subjects/order-events-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"OrderEvent\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"status\",\"type\":\"string\"}]}"
  }'
# 返回：{"id": 1}

查询 Schema

# 查询所有 Subject
curl http://localhost:8081/subjects

# 查询某 Subject 的所有版本号
curl http://localhost:8081/subjects/order-events-value/versions

# 查询最新版本的 Schema 详情（含 id、version、schema 字段）
curl http://localhost:8081/subjects/order-events-value/versions/latest

# 按 Schema ID 查询（跨 Subject 通用）
curl http://localhost:8081/schemas/ids/1

兼容性检查

# 检查新 Schema 是否与当前最新版本兼容（提交前预检）
curl -X POST \
  http://localhost:8081/compatibility/subjects/order-events-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"OrderEvent\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"note\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}'
# 返回：{"is_compatible": true}

修改兼容性级别

# 修改全局兼容性级别
curl -X PUT http://localhost:8081/config \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL"}'

# 修改单个 Subject 的兼容性级别（覆盖全局设置）
curl -X PUT http://localhost:8081/config/order-events-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "BACKWARD"}'

删除 Schema

# 软删除一个 Subject（数据保留，标记 deleted=true）
curl -X DELETE http://localhost:8081/subjects/order-events-value

# 硬删除（永久不可恢复，需先软删除）
curl -X DELETE "http://localhost:8081/subjects/order-events-value?permanent=true"

# 软删除某个特定版本
curl -X DELETE http://localhost:8081/subjects/order-events-value/versions/1

---

多租户：Subject 命名策略

Subject 是 Schema Registry 中的命名空间单位，有三种内置策略：

策略	Subject 命名规则	适用场景
TopicNameStrategy（默认）	<topic>-key / <topic>-value	一个 Topic 只承载一种 Record 类型
RecordNameStrategy	<namespace>.<RecordName>	同一 Schema 被多个 Topic 复用
TopicRecordNameStrategy	<topic>-<namespace>.<RecordName>	一个 Topic 包含多种 Record 类型

配置生产者使用非默认策略：

// 切换为 RecordNameStrategy（按 Schema 的全限定名注册）
props.put("value.subject.name.strategy",
        "io.confluent.kafka.serializers.subject.RecordNameStrategy");

---

掌握检查

• [ ] 能说出 Schema Registry 5 字节头部格式（Magic Byte + Schema ID）的作用
• [ ] 理解 BACKWARD / FORWARD / FULL / NONE 四种兼容性的差异
• [ ] 能编写包含 nullable 字段、嵌套 record、enum 和 map 的 Avro Schema
• [ ] 能配置 Java 生产者/消费者使用 KafkaAvroSerializer / KafkaAvroDeserializer
• [ ] 能用 REST API 完成注册、查询、兼容性检查、删除操作
• [ ] 理解三种 Subject 命名策略的适用场景

练习题

1. 设计一个 PaymentEvent Avro Schema，包含：支付 ID、订单 ID、金额（decimal）、支付方式（enum: ALIPAY/WECHAT/CARD）、失败原因（nullable string），确保向 BACKWARD 兼容演进时可以安全新增字段。
2. 当前 Schema 的某字段 phone 类型是 string，你需要改为 ["null", "string"]，请说明这个修改在 BACKWARD 模式下是否兼容，并给出验证的 REST API 命令。
3. 生产者使用 v2 Schema 写入消息时，分别分析本文 GenericRecord 消费者与使用 v1 SpecificRecord 的消费者如何反序列化。说明为什么“只注册了 v1”不必然导致报错，并分析 reader/writer Schema 不兼容时可能出现的错误及避免方法。

学习连接

前置知识: Kafka Connect - 连接器生态 下一步: KSQL 流式 SQL - 流式数据查询 相关概念: 数据治理、Avro 序列化原理

---