Kafka MQTT源转JDBC Sink至PostgreSQL报错解决咨询

错误根源 #

你配置的InsertField使用了静态字段插入逻辑（static.field+static.value），但实际需求是将Kafka消息的Key（即MQTT原始topic）存入数据库topic字段，而非固定静态值。这种配置方向的错误直接导致了参数解析异常（提示static.value为null，本质是静态配置逻辑与实际需求冲突）。同时还存在两个辅助问题：

• 原value.converter为ByteArrayConverter，JDBC Sink无法直接将字节数组映射为PostgreSQL的浮点数字段；
• 原TimestampConverter配置用于转换已有时间字段，但你需要的是自动插入当前时间戳。

修正后的JDBC Sink Connector配置 #

{
  "name": "jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": 1,
    "topics": "mqtt_messages",
    "connection.url": "jdbc:postgresql://timescaledb:5432/metrics",
    "connection.user": "postgres",
    "connection.password": "password",
    "auto.create": true,
    "auto.evolve": true,
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "topic",
    "table.name.format": "power_metrics",
    "fields.whitelist": "topic,value,time",
    "transforms": "InsertTopic,ConvertValue,InsertTime",
    // 从消息Key提取MQTT原始topic，插入到Value的topic字段
    "transforms.InsertTopic.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertTopic.source.field": "__key",
    "transforms.InsertTopic.field": "topic",
    // 将字符串格式的消息值转换为浮点数
    "transforms.ConvertValue.type": "org.apache.kafka.connect.transforms.Cast$Value",
    "transforms.ConvertValue.spec": "value:float64",
    // 自动插入当前时间戳到time字段
    "transforms.InsertTime.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertTime.timestamp.field": "time",
    // 配置转换器：Key为字符串，Value先转为字符串再处理
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}

关键修正点说明 #

1. 替换静态字段插入逻辑：用InsertField的source.field=__key从消息Key中提取MQTT原始topic，存入Value的topic字段，匹配数据库表需求；
2. 调整值转换器与类型转换：将value.converter改为StringConverter，配合Cast转换把字符串格式的功率值转为浮点数，适配数据库的value字段类型；
3. 简化时间戳插入：用InsertField的timestamp.field自动插入当前时间戳，替代原有的TimestampConverter（原配置用于转换已有时间字段，不符合需求）；
4. 更新字段白名单：添加time字段，确保所有数据库表字段都被包含。

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