使用MS SQL Server Sink Connector同步Kafka数据至数据库失败排查

核心问题 #

从DLQ的错误信息Unknown magic byte!可以明确，消息序列化格式与连接器的反序列化要求不匹配：

• 你配置的MS SQL Server Sink Connector使用了io.confluent.connect.json.JsonSchemaConverter，该转换器要求消息必须是Confluent Schema Registry兼容的JSON Schema序列化格式（包含固定的magic byte前缀、schema ID等元数据）。
• 手动通过Confluent Cloud生产消息时，平台自动处理了符合要求的序列化；但Postman、Java应用、Boomi等工具发送的是纯JSON字符串，缺少Schema Registry序列化格式的头部元数据，导致连接器无法解析。

疑问解答 #

1. MS SQL Server Sink Connector对第三方工具是否有限制？
   没有限制。问题不在连接器本身，而是第三方工具发送的消息格式不符合连接器配置的反序列化规则。只要消息格式匹配，任何工具生产的消息都能被正常处理。

2. 可行的解决方法？
   提供两种方向的解决方案，根据你的业务需求选择：

方案1：调整生产者，发送符合Schema Registry格式的消息（推荐） #

这是Confluent生态的标准用法，能利用Schema Registry的schema验证、版本管理能力：

• Java应用：确保使用Confluent官方的KafkaJsonSchemaSerializer，并配置Schema Registry地址：

  Properties props = new Properties();
  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-bootstrap-servers");
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaJsonSchemaSerializer.class);
  props.put("schema.registry.url", "your-schema-registry-url");
  // 其他生产者配置...
  

• Postman：
  • 先通过Schema Registry API注册你的消息schema；
  • 按照Confluent JSON Schema消息格式构造请求：消息前添加1字节的magic byte（值为0），再添加4字节的schema ID（大端序），最后拼接JSON payload；
  • 或者直接使用Confluent Cloud的REST Proxy发送消息，REST Proxy会自动处理序列化逻辑。
• Boomi：
  在Boomi的Kafka连接器配置中，选择Confluent兼容的JSON Schema序列化方式，或通过自定义脚本生成符合格式的消息字节流。

方案2：调整连接器配置，兼容纯JSON消息 #

如果不需要Schema Registry的能力，可以修改连接器的转换器配置，让它直接解析纯JSON：

• 将连接器的value.converter改为org.apache.kafka.connect.json.JsonConverter；
• 添加配置value.converter.schemas.enable=false（如果你的消息本身不包含schema字段）；
• 重启连接器后，即可处理纯JSON格式的消息。

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