MQTT源连接器ByteArray值转换失败，如何配置转换器实现转String处理

一、基础转换配置（ByteArray转UTF-8字符串） #

你之前修改value.converter未生效的核心原因有两个：

1. 未显式指定StringConverter的编码规则，且保留了不需要的schema.enable配置
2. 分布式模式下转换器配置必须写在连接器的config参数中，仅修改worker配置文件不生效

修正后的基础配置如下：

{
   "name":"MqttSourceConnector",
   "config":{
      "connector.class":"io.confluent.connect.mqtt.MqttSourceConnector",
      "tasks.max":"2",
      "key.converter":"org.apache.kafka.connect.storage.StringConverter",
      "key.converter.schemas.enable":"false",
      // 替换ByteArrayConverter为StringConverter，显式指定编码
      "value.converter":"org.apache.kafka.connect.storage.StringConverter",
      "value.converter.encoding":"UTF-8",
      "value.converter.schemas.enable":"false",
      "mqtt.server.uri":"ssl://XXXXXX.XXXX.amazonaws.com:1883",
      "mqtt.topics":"/mqtt/topic",
      "mqtt.ssl.trust.store.path":"XXXXXXXXXX.jks",
      "mqtt.ssl.trust.store.password":"XXXXX",
      "mqtt.ssl.key.store.path":"XXXXXXXXX.jks",
      "mqtt.ssl.key.store.password":"XXXXXX",
      "mqtt.ssl.key.password":"XXXXXXXX",
      "max.retry.time.ms":86400000,
      "mqtt.connect.timeout.seconds":2,
      "mqtt.keepalive.interval.seconds":4
  }
}

应用该配置后，Kafka中存储的消息值将直接为UTF-8编码的字符串，不再是字节数组。

二、进阶配置：JSON解析+动态Topic/Key提取 #

如果需要从MQTT payload中提取字段作为Kafka的Topic和Key，可使用Kafka Connect自带的单消息转换（SMT）能力实现，注意需要先调整Python侧的输出为标准双引号JSON格式，不要使用Python原生单引号字典，示例：

import json
# 正确写法，生成标准JSON字符串
payload = json.dumps({"id": 42, "cost": 4000, "kafka_topic": "device_bill", "msg_key": "device_001"})

添加SMT后的完整配置如下：

{
   "name":"MqttSourceConnector",
   "config":{
      "connector.class":"io.confluent.connect.mqtt.MqttSourceConnector",
      "tasks.max":"2",
      "key.converter":"org.apache.kafka.connect.storage.StringConverter",
      "key.converter.schemas.enable":"false",
      "value.converter":"org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable":"false",
      // 原有MQTT连接配置保留不变
      "mqtt.server.uri":"ssl://XXXXXX.XXXX.amazonaws.com:1883",
      "mqtt.topics":"/mqtt/topic",
      "mqtt.ssl.trust.store.path":"XXXXXXXXXX.jks",
      "mqtt.ssl.trust.store.password":"XXXXX",
      "mqtt.ssl.key.store.path":"XXXXXXXXX.jks",
      "mqtt.ssl.key.store.password":"XXXXXX",
      "mqtt.ssl.key.password":"XXXXXXXX",
      "max.retry.time.ms":86400000,
      "mqtt.connect.timeout.seconds":2,
      "mqtt.keepalive.interval.seconds":4,
      // 新增SMT配置
      "transforms":"parseJson,extractKey,routeTopic",
      // 第一步：将JSON字符串解析为结构化对象
      "transforms.parseJson.type":"org.apache.kafka.connect.transforms.JsonToStruct",
      "transforms.parseJson.schema.enable":"false",
      // 第二步：提取payload中指定字段作为Kafka消息Key，示例取id字段
      "transforms.extractKey.type":"org.apache.kafka.connect.transforms.ExtractField$Value",
      "transforms.extractKey.field":"id",
      // 第三步：提取payload中指定字段作为Kafka目标Topic，示例取kafka_topic字段
      "transforms.routeTopic.type":"io.confluent.connect.transforms.ExtractTopic$Value",
      "transforms.routeTopic.field":"kafka_topic"
  }
}

三、验证说明 #

配置更新后重启连接器，清空原有异常数据的Topic再进行测试，即可实现：

• Kafka消息值为解析后的标准JSON格式
• 消息Key为payload中指定的字段值
• 消息自动路由到payload指定的Kafka Topic

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