使用Kafka-Connect与Schema Registry反序列化Protobuf数据失败求助

错误核心原因 #

错误栈中的Unknown magic byte!和Error deserializing Protobuf message for id -1是关键线索：

• Confluent官方的Protobuf序列化器（io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer）会在每条消息头部添加1字节固定magic值（0）和4字节Schema Registry中的schema ID，用于和Registry交互获取对应解析规则。
• 你的Kafka Topic中的消息没有这个前缀，是直接序列化的原始Protobuf二进制数据，导致Connect的ProtobufConverter无法识别消息格式，触发反序列化失败。

修复方案 #

根据你的实际场景选择以下一种方式：

方案1：修改生产者，使用Confluent Protobuf序列化器 #

如果可以调整生产者代码/配置，让生产者用Confluent的序列化器发送消息，这样会自动处理schema注册和消息前缀：
生产者配置需添加：

value.serializer=io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
schema.registry.url=http://localhost:8081

配置后，生产者会自动将schema注册到Registry（未注册时），并给消息加上标准前缀，Connect即可正常解析。

方案2：修改Connect配置，直接解析原始Protobuf数据 #

如果无法修改生产者，可让Connect跳过Schema Registry，直接指定Protobuf消息类型解析原始二进制数据：

1. 移除value.converter.schema.registry.url配置项
2. 添加value.converter.protobuf.message.type，指定Protobuf消息的全限定类名（例如你的nilian.Person）
3. 确保Kafka Connect的类路径中包含该Protobuf类编译后的jar包

修改后的连接器配置示例：

{
   "name":"proto_sink",
   "config":{
      "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
      "connection.password":"some_password",
      "topics":"some_topic",
      "key.converter.schemas.enable":"false",
      "auto.evolve":"true",
      "connection.user":"some_user",
      "value.converter.schemas.enable":"false",
      "name":"sink_proto",
      "auto.create":"true",
      "connection.url":"jdbc:postgresql://localhost:5432/some_db",
      "value.converter":"io.confluent.connect.protobuf.ProtobufConverter",
      "value.converter.protobuf.message.type":"nilian.Person",
      "insert.mode":"insert",
      "key.converter":"org.apache.kafka.connect.storage.StringConverter"
   }
}

验证步骤 #

你可以用以下命令查看Topic中消息的原始格式，确认是否带Confluent标准前缀：

kafka-console-consumer --bootstrap-server localhost:9092 --topic some_topic --from-beginning --property print.key=true --property print.value=true --property value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

如果输出为纯乱码（无固定前缀特征），即可确认是原始Protobuf数据，按上述方案修复即可。

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