Confluent Kafka Avro生产者Schema解析与注册冲突问题咨询

1. 未知named schema报错 #

你分开存储两个有依赖关系的avsc文件时，单独加载Subscription.avsc的过程中，本地Avro解析器无法识别你引用的io.codebrews.schema.test.Keys类型——这个解析过程是本地完成的，和你是否提前把Keys schema注册到Schema Registry没有关系，本地解析时没有加载过Keys的schema定义，就会直接抛出找不到对应schema的错误。

2. 409 Schema不兼容报错 #

Schema Registry默认启用兼容性校验（默认策略为BACKWARD），你手动修改schema内容注册后，新版本和该主题下已存在的历史schema版本不符合兼容性规则，就会返回409错误。

schema引用问题修复 #

两种可选方案：

• 直接把Keys schema的定义内嵌到Subscription.avsc的types字段中，不需要拆分两个文件，修改后的Subscription.avsc示例：

{
    "namespace": "io.codebrews.schema.test",
    "type": "record",
    "name": "Subscription",
    "types": [
        {
            "name": "Keys",
            "type": "record",
            "fields": [
                {
                    "name": "name",
                    "type": "string"
                },
                {
                    "name": "email",
                    "type": "string"
                }
            ]
        }
    ],
    "fields": [
        {
            "name": "test",
            "type": "string"
        },
        {
            "name": "keys",
            "type": "Keys"
        }
    ]
}

• 保留两个文件，加载Subscription前先加载Keys schema到本地解析上下文：

def load_avro_schema_from_file(schema_file):
    key_schema_string = """{"type": "string"}"""
    key_schema = avro.loads(key_schema_string)
    # 先加载依赖的Keys schema
    keys_schema = avro.load("./avro/Keys.avsc")
    # 再加载主schema，此时解析器可以识别已经加载过的Keys类型
    value_schema = avro.load("./avro/" + schema_file)
    return key_schema, value_schema

409错误修复 #

• 测试环境可直接删除test-value主题对应的旧schema版本，重新注册即可
• 生产环境可根据业务需要调整Schema Registry对应subject的兼容性策略，或者修改schema内容符合现有兼容性规则

直接在你的生产者配置producer_config中添加配置项即可：

producer_config = {
    # 你原有其他配置
    "auto.register.schemas": False
}

配置生效后，AvroProducer不会再自动把本地schema注册到Schema Registry，只会使用Registry中已经存在的对应schema。

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