Python解码Kafka中Avro格式数据时遭遇AssertionError问题

问题情况 #

使用Python 3.11.6解码由Apicurio编码的Kafka Avro格式数据时触发AssertionError，数据由Debezium Oracle连接器生成，采用Envelope结构，当前代码通过手动跳过前5字节尝试解码。

错误信息 #

AssertionError: 解码字节流 b'\xf0@\x02 ABCRequest\x161.9.7.Final\x0cabc-xyz-xyz.abc.com\xb0\xe4\xdb\xd0\xf2b\x00\x08true\x0esystem\x00\x10abccard\ntablename\x00\x02\x12601987152\x00\x00\x00\x02\x00\x00\x02r\x02\xac\x90\xc9\xbf\xf2b\x00' 时触发断言错误

代码问题分析 #

1. 手动跳过字节的风险：当前代码用message_bytes.seek(5)跳过前5字节，默认适配Confluent格式的魔术字节+Schema ID长度，但Apicurio的编码格式可能存在差异，或实际消息前缀长度不匹配，导致后续字节流解析错位。
2. 硬编码Schema不一致：本地硬编码的Schema可能与Apicurio Registry中存储的实际Schema版本不匹配，字段类型、默认值或嵌套结构的细微差异都会触发解码断言错误。
3. 未利用官方解码工具：直接使用avro.io基础API手动解码，未借助Schema Registry客户端自动匹配Schema的能力，易出现格式兼容问题。

解决方案 #

方案1：使用confluent-kafka-avro库自动解码 #

利用CachedSchemaRegistryClient和AvroConsumer自动从Registry拉取对应Schema并解码，避免手动处理字节流：

from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient
from confluent_kafka import KafkaError

def main():
    print("Listening *****************")
    SCHEME_REGISTERY = "http://localhost:8082"
    TOPIC = "abc"
    BROKER = "localhost:9092"
    
    schema_registry_client = CachedSchemaRegistryClient({
        'url': SCHEME_REGISTERY
    })

    consumer = AvroConsumer({
        'bootstrap.servers': BROKER,
        'group.id': 'avro-consumer-group',
        'auto.offset.reset': 'latest',
        'enable.auto.commit': False
    }, schema_registry_client=schema_registry_client)

    consumer.subscribe([TOPIC])

    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    print(msg.error())
                    break
            print("\n")
            print("解码结果:", msg.value())
            print("\n")
    finally:
        consumer.close()

if __name__ == "__main__":
    main()

方案2：修正手动解码的字节处理逻辑 #

若必须手动解码，先正确解析消息前缀的Schema ID，再从Registry拉取对应Schema：

import struct
from confluent_kafka.schema_registry import SchemaRegistryClient
from avro.io import DatumReader, BinaryDecoder
import avro.schema
import io
from kafka import KafkaConsumer

def get_schema_from_registry(schema_id, registry_url):
    client = SchemaRegistryClient({'url': registry_url})
    schema_str = client.get_schema(schema_id).schema_str
    return avro.schema.Parse(schema_str)

def decode_avro_message(msg_value, registry_url):
    # 解析前缀：魔术字节(1字节) + Schema ID(4字节，大端)
    magic_byte, schema_id = struct.unpack('>bI', msg_value[:5])
    assert magic_byte == 0, "无效的Avro消息魔术字节"
    
    # 从Registry拉取对应Schema
    schema = get_schema_from_registry(schema_id, registry_url)
    reader = DatumReader(schema)
    
    # 解码剩余字节
    message_bytes = io.BytesIO(msg_value[5:])
    decoder = BinaryDecoder(message_bytes)
    return reader.read(decoder)

def main():
    print("Listening *****************")
    SCHEME_REGISTERY = "http://localhost:8082"
    TOPIC = "abc"
    BROKER = "localhost:9092"

    consumer = KafkaConsumer(
        TOPIC,
        bootstrap_servers=[BROKER],
        auto_offset_reset='latest',
        enable_auto_commit=False
    )

    for msg in consumer:
        msg_value = msg.value
        print("\n")
        try:
            result = decode_avro_message(msg_value, SCHEME_REGISTERY)
            print("解码结果:", result)
        except Exception as e:
            print("解码失败:", str(e))
        print("\n")

main()

额外检查项 #

• 确认Apicurio Registry地址SCHEME_REGISTERY配置正确，Python服务可正常访问该地址。
• 验证Registry中对应主题的Schema与本地硬编码Schema完全一致，重点检查嵌套结构、字段类型和默认值。
• 检查Debezium连接器配置，确保Avro编码格式与Schema Registry兼容。

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