FULL兼容模式下添加Avro可选字段遇Schema Registry兼容性问题

问题场景 #

初始注册的Avro Schema包含两个顶级Record：

[{"type":"record","namespace":"kafka.avro.personal.schema.nested","name":"Address","fields":[{"name":"streetaddress","type":"string"},{"name":"city","type":"string"}]},{"type":"record","name":"Personal","namespace":"kafka.avro.personal.schema.nested","fields":[{"name":"firstname","type":"string"},{"name":"lastname","type":"string"},{"name":"status","default":"SALARY","type":{"type":"enum","name":"Status","symbols":["RETIRED","SALARY","HOURLY","PART_TIME"]}},{"name":"address","type":["null","Address"]}]}]

注册到Schema Registry后，给Personal Record添加一个带默认值的可选字段：

{"name":"optionalT","type":["null","string"],"default":null}

但使用FULL兼容模式重新发布消息时，仍触发兼容性错误，错误码409：

Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error registering Avro schema:{"type":"record","name":"Personal","namespace":"kafka.avro.personal.schema.nested","fields":[{"name":"firstname","type":"string"},{"name":"lastname","type":"string"},{"name":"status","type":{"type":"enum","name":"Status","symbols":["RETIRED","SALARY","HOURLY","PART_TIME"]},"default":"SALARY"},{"name":"address","type":["null",{"type":"record","name":"Address","fields":[{"name":"streetaddress","type":"string"},{"name":"city","type":"string"}]}]},{"name":"optionalT","type":["null","string"],"default":null}]}
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:103)
    ...
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema being registered is incompatible with an earlier schema; error code:409
    ...

原因分析 #

FULL兼容模式要求新旧Schema双向兼容（旧Schema可读新消息，新Schema也可读旧消息），问题出在新Schema对Address类型的定义方式：

• 初始Schema中Address是独立的顶级Record，Personal的address字段直接引用该顶级类型；
• 新Schema中把Address内联到了address字段的类型定义里，而非引用已注册的顶级Address；
• Schema Registry会将内联定义的Address视为全新类型（即使名称、结构完全一致），导致新旧Schema中address字段的类型本质不匹配，触发兼容性检查失败。

解决方案 #

修改新Schema，保持对已注册顶级Address类型的引用，不要内联定义：

方案1：仅提交修改后的Personal Schema（引用已存在的Address） #

{"type":"record","name":"Personal","namespace":"kafka.avro.personal.schema.nested","fields":[{"name":"firstname","type":"string"},{"name":"lastname","type":"string"},{"name":"status","type":{"type":"enum","name":"Status","symbols":["RETIRED","SALARY","HOURLY","PART_TIME"]},"default":"SALARY"},{"name":"address","type":["null","Address"]},{"name":"optionalT","type":["null","string"],"default":null}]}

方案2：提交完整的顶级Schema集合（保留Address的独立定义） #

[{"type":"record","namespace":"kafka.avro.personal.schema.nested","name":"Address","fields":[{"name":"streetaddress","type":"string"},{"name":"city","type":"string"}]},{"type":"record","name":"Personal","namespace":"kafka.avro.personal.schema.nested","fields":[{"name":"firstname","type":"string"},{"name":"lastname","type":"string"},{"name":"status","default":"SALARY","type":{"type":"enum","name":"Status","symbols":["RETIRED","SALARY","HOURLY","PART_TIME"]}},{"name":"address","type":["null","Address"]},{"name":"optionalT","type":["null","string"],"default":null}]}]

使用上述任一修改后的Schema重新注册，即可通过FULL兼容模式的检查，成功发布消息。

内容的提问来源于stack exchange，提问作者user10523689