Kafka正常收发Avro数据时使用JDBC Sink Connector报Unknown magic byte错误

• Confluent Avro序列化后的消息会在头部固定写入一个值为0x00的魔术字节，用于和其他序列化格式做区分，反序列化时会先校验该字节是否符合预期
• 你自行实现的Java消费者显式配置了KafkaAvroDeserializer作为值反序列化器，同时配置了正确的Schema Registry地址，因此可以正常解析Avro格式消息
• 你提交的JDBC Sink Connector配置中未指定Avro格式对应的转换器，默认反序列化器无法识别魔术字节，直接抛出SerializationException: Unknown magic byte!错误

修改Connector配置，添加和生产者序列化逻辑对齐的转换器配置，完整的部署请求如下：

curl -XPOST --header "Content-Type: application/json"  XXXXXXXXXX.XXXXXXXXXX.com:8083/connectors  -d 
'{  
    "name": "sink_AVROTESTNEW",  
    "config": {    
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",    
        "tasks.max": 3,   
        "connection.url": "jdbc:oracle:thin:@XXXXXXXXXX:1854:XXXXXXXXXX",
        "connection.user": "XXXXXXXXXX",    
        "connection.password": "XXXXXXXXXX",    
        "table.name.format": "AVROTEST",
        "topics": "AVROTESTNEW",
        "auto.create": "false",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "http://XXXXXXXXXX.XXXXXXXX.XXXXXX:8071/",
        "value.converter.schemas.enable": "true"
    }
}'

额外排查项 #

如果修改配置后仍报错，可按以下顺序检查：

• 确认目标TopicAVROTESTNEW中没有非Confluent Avro格式的历史脏数据，若有可清空Topic后重新发送测试消息验证
• 确认Kafka Connect集群节点可以正常访问Schema Registry的8071端口，无网络策略、防火墙限制
• 检查Oracle的AVROTEST表字段名、字段类型是否和Avro Schema中定义的字段完全匹配，避免后续写入时报字段不匹配错误

内容的提问来源于stack exchange，提问作者Bünyamin Şentürk