confluent kafka producer avro schema error ClientError: Schema parse failed: Unknown named schema

Question

我在 kafka 的生产者方面工作，以在主题中推送消息。 我正在使用confluent-kafkaavro producer。

Liked issue on github

以下是我的架构.avsc 文件。

Keys.avsc

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

Test.avsc

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

Producer.py

key_schema, value_schema = load_avro_schema_from_file('Subscription.avsc')

try:
    producer = avro.AvroProducer(producer_config, default_key_schema=key_schema, default_value_schema=value_schema)
except Exception as e:
    raise e

def load_avro_schema_from_file(schema_file):
    key_schema_string = """
    {"type": "string"}
    """

    key_schema = avro.loads(key_schema_string)
    value_schema = avro.load("./avro/" + schema_file)

    return key_schema, value_schema

当我尝试注册Keys.avsc 时，它工作正常，没有错误。但是当我在注册Keys.avsc 后尝试注册Test.avsc 时。我得到以下错误。

confluent_kafka.avro.error.ClientError：架构解析失败：未知命名架构“io.codebrews.schema.test.Keys”，已知名称：['io.codebrews.schema.test.Subscription']。

手动注册架构后。

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

在我的主题中推送消息时，出现以下错误。

ClientError: Incompatible Avro schema:409 message:{'error_code': 409, 'message': '正在注册的模式与主题“test-value”的早期模式不兼容。

我在这里做错了吗？

还有谁能帮我如何在 python 中停止自动模式注册？

Answer 1

错误与解析器有关，与注册表注册无关...

您的 AVSC 文件需要完全包含所有记录类型。当解析器读取一个文件时，它没有办法知道其他文件

如果您从 AVDL 文件开始，然后将其转换为 AVSC，则记录会正确嵌入到外部记录中。

具体来说，

"fields" :[{
  "name": "keys",
  "type": {
       "type": "record", 
       "namespace": "io.codebrews.schema.test.Keys", 
       ... 
  } 
}