1
// Schema V1
2
{
3
  "type": "record",
4
  "name": "Order",
5
  "fields": [
6
    {"name": "orderId", "type": "string"},
7
    {"name": "amount", "type": "double"},
8
    {"name": "status", "type": "string", "default": "CREATED"}
9
  ]
10
}
11

12
// Schema V2（向后兼容：新增字段有默认值）
13
{
14
  "type": "record",
15
  "name": "Order",
16
  "fields": [
17
    {"name": "orderId", "type": "string"},
18
    {"name": "amount", "type": "double"},
19
    {"name": "status", "type": "string", "default": "CREATED"},
20
    {"name": "currency", "type": "string", "default": "CNY"},
21
    {"name": "items", "type": {"type": "array", "items": "string"}, "default": []}
22
  ]
23
}

1
// Schema V1
2
message Order {
3
  string order_id = 1;
4
  double amount = 2;
5
  string status = 3;
6
}
7

8
// Schema V2（兼容：新增字段，不修改已有编号）
9
message Order {
10
  string order_id = 1;
11
  double amount = 2;
12
  string status = 3;
13
  string currency = 4;    // 新增字段，新编号
14
  repeated string items = 5;  // 新增字段，新编号
15
  // 注意：不要复用已删除字段的编号！
16
  // reserved 6, 7;  // 标记已删除的编号
17
}

1
# 注册 OrderEvent V1 Schema
2
curl -X POST http://localhost:8081/subjects/order-event/versions   -H "Content-Type: application/vnd.schemaregistry.v1+json"   -d '{
3
    "schema": "{"type":"record","name":"OrderEvent","fields":[{"name":"orderId","type":"string"},{"name":"amount","type":"double"}]}"
4
  }'
5
# 返回: {"id":1}
6

7
# 注册 V2 — 新增可选字段（BACKWARD 兼容）
8
curl -X POST http://localhost:8081/subjects/order-event/versions   -H "Content-Type: application/vnd.schemaregistry.v1+json"   -d '{
9
    "schema": "{"type":"record","name":"OrderEvent","fields":[{"name":"orderId","type":"string"},{"name":"amount","type":"double"},{"name":"currency","type":["null","string"],"default":null}]}"
10
  }'
11
# 返回: {"id":2}
12

13
# 查看所有版本
14
curl http://localhost:8081/subjects/order-event/versions
15
# 返回: [1, 2]
16

17
# 检查兼容性（注册前预检）
18
curl -X POST http://localhost:8081/compatibility/subjects/order-event/versions/latest   -H "Content-Type: application/vnd.schemaregistry.v1+json"   -d '{"schema": "...新 Schema..."}'
19
# 返回: {"is_compatible":true}

1
# 启动 Schema Registry
2
docker run -d --name schema-registry \
3
    -p 8081:8081 \
4
    -e SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=localhost:9092 \
5
    -e SCHEMA_REGISTRY_HOST_NAME=localhost \
6
    confluentinc/cp-schema-registry
7

8
# 注册 Schema
9
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
10
    --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}]}"}' \
11
    http://localhost:8081/subjects/order-value/versions
12

13
# 查看所有 Schema
14
curl http://localhost:8081/subjects
15

16
# 查看 Schema 版本
17
curl http://localhost:8081/subjects/order-value/versions
18

19
# 查看特定版本
20
curl http://localhost:8081/subjects/order-value/versions/1
21

22
# 检查兼容性
23
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
24
    --data '{"schema": "...new schema..."}' \
25
    http://localhost:8081/compatibility/subjects/order-value/versions/latest

1
# 多实例部署（通过 Kafka Topic 实现一致性）
2
# Schema Registry 使用 Kafka Topic "_schemas" 存储所有 Schema
3
# 多实例通过 Kafka 的消费者协调机制保证 Leader 选举和数据一致
4

5
# 实例 1
6
docker run -d --name schema-registry-1 \
7
    -p 8081:8081 \
8
    -e SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=kafka:9092 \
9
    -e SCHEMA_REGISTRY_HOST_NAME=schema-registry-1 \
10
    -e SCHEMA_REGISTRY_LISTENERS=http://0.0.0.0:8081 \
11
    confluentinc/cp-schema-registry
12

13
# 实例 2
14
docker run -d --name schema-registry-2 \
15
    -p 8082:8081 \
16
    -e SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=kafka:9092 \
17
    -e SCHEMA_REGISTRY_HOST_NAME=schema-registry-2 \
18
    -e SCHEMA_REGISTRY_LISTENERS=http://0.0.0.0:8081 \
19
    confluentinc/cp-schema-registry

1
// 配置 Subject Name Strategy（Producer 端）
2
props.put("value.subject.name.strategy",
3
    TopicRecordNameStrategy.class.getName());
4

5
// 效果：同一 Topic 可以包含多种 Avro Record
6
// orders Topic 中可以同时包含 Order、Payment、Shipment 三种消息
7
// 每种消息有独立的 Schema 版本和兼容性策略

1
// Producer：使用 Schema Registry
2
Properties props = new Properties();
3
props.put("bootstrap.servers", "localhost:9092");
4
props.put("key.serializer", StringSerializer.class.getName());
5
props.put("value.serializer", KafkaAvroSerializer.class.getName());
6
props.put("schema.registry.url", "http://localhost:8081");
7

8
KafkaProducer<String, Order> producer = new KafkaProducer<>(props);
9

10
// 发送 Avro 消息（自动注册 Schema）
11
Order order = Order.newBuilder()
12
    .setOrderId("ORD-001")
13
    .setAmount(99.9)
14
    .setStatus("CREATED")
15
    .build();
16
producer.send(new ProducerRecord<>("orders", order.getOrderId(), order));
17

18
// Consumer：使用 Schema Registry
19
props.put("key.deserializer", StringDeserializer.class.getName());
20
props.put("value.deserializer", KafkaAvroDeserializer.class.getName());
21
props.put("schema.registry.url", "http://localhost:8081");
22
props.put("specific.avro.reader", "true");  // 使用生成的 Avro 类
23

24
KafkaConsumer<String, Order> consumer = new KafkaConsumer<>(props);
25
consumer.subscribe(List.of("orders"));
26

27
while (true) {
28
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(100));
29
    for (var record : records) {
30
        Order o = record.value();  // 自动反序列化
31
        System.out.println(o.getOrderId() + ": " + o.getAmount());
32
    }
33
}

1
# 设置 Topic 级别的兼容性
2
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
3
    --data '{"compatibility": "BACKWARD"}' \
4
    http://localhost:8081/config/order-value
5

6
# 可选值：NONE, BACKWARD, BACKWARD_TRANSITIVE, FORWARD, FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE

1
// V1：初始订单 Schema（Avro）
2
@AvroSchema("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[...]}")
3
public record OrderV1(String orderId, double amount, String status) {}
4

5
// V2：新增 currency 字段（向后兼容）
6
public record OrderV2(String orderId, double amount, String status, String currency) {
7
    public OrderV2(String orderId, double amount, String status) {
8
        this(orderId, amount, status, "CNY");  // 默认值
9
    }
10
}
11

12
// V3：新增 items 字段（向后兼容）
13
public record OrderV3(String orderId, double amount, String status,
14
                       String currency, List<String> items) {
15
    public OrderV3(String orderId, double amount, String status, String currency) {
16
        this(orderId, amount, status, currency, List.of());  // 默认空列表
17
    }
18
}

1
// 不兼容变更：amount 从 double 改为 BigDecimal
2
// 解决方案：新增字段，旧字段标记废弃
3

4
// V4：新增 preciseAmount，保留 amount
5
public record OrderV4(
6
    String orderId,
7
    @Deprecated double amount,          // 保留，标记废弃
8
    String status,
9
    String currency,
10
    List<String> items,
11
    String preciseAmount                 // 新增：精确金额字符串
12
) {}
13

14
// 消费者端：优先使用 preciseAmount
15
public class OrderConsumer {
16
    public void process(OrderV4 order) {
17
        BigDecimal amount = order.preciseAmount() != null
18
            ? new BigDecimal(order.preciseAmount())
19
            : BigDecimal.valueOf(order.amount());  // 降级到旧字段
20
        // 处理订单...
21
    }
22
}

1
// 错误示范：删除必填字段
2
// V1: {"name": "orderId", "type": "string"}, {"name": "amount", "type": "double"}
3
// V2: {"name": "amount", "type": "double"}  ← 删除了 orderId，BACKWARD 不兼容！
4

5
// 正确做法：先设为可选
6
// V2: {"name": "orderId", "type": ["null", "string"], "default": null}, {"name": "amount", "type": "double"}
7
// V3（等所有 Consumer 更新后）: 删除 orderId