Confluent Kafka Timestamp Converter时间转换值偏差问题求助

环境与场景 #

• Apache Kafka + Confluent Connect v7.3.2
• 部署MongoDB Sink Connector，将包含2个时间戳字段（格式yyyy-MM-dd'T'HH:mm:ss.SSSSS，小数点后5位）的消息写入MongoDB集合
• 使用TimestampConverter SMT将字符串时间转换为MongoDB支持的ISODate类型（仅支持3位毫秒精度）

关键配置 #

name=mongo-sink
topics=topic-with-sink
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

connection.uri=mongodb://192.168.41.3:27017
database=MarketData
collection=Mycollection

# 仅启用LastUpdated字段的时间转换
transforms=TimestampToIsoDate
transforms.TimestampToIsoDate.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.TimestampToIsoDate.target.type=Timestamp
transforms.TimestampToIsoDate.field=LastUpdated
transforms.TimestampToIsoDate.format=yyyy-MM-dd'T'HH:mm:ss.SSSSS

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.BsonOidStrategy
post.processor.chain=com.mongodb.kafka.connect.sink.processor.DocumentIdAdder

writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneDefaultStrategy

异常现象 #

Kafka原消息中的时间字段：

"LastUpdated":"2023-08-22T13:52:50.13426"

MongoDB中存储的结果：

"lastUpdated": "2023-08-22T13:53:03.426Z"

两者时间偏差3秒，部分场景偏差可达数分钟，已排除时区差异问题。

排查与修复方案 #

1. 核心原因：时间格式解析逻辑错误 #

Confluent Connect v7.3.2的TimestampConverter SMT底层依赖SimpleDateFormat，而该类对格式符S的定义是毫秒（最多3位）：

• 当传入5位数字13426时，会被直接当作毫秒数（13426ms = 13秒426毫秒），而非“134毫秒+26微秒”
• 原时间13:52:50.13426被错误计算为13:52:50 + 13.426秒 = 13:53:03.426，与观测到的偏差完全匹配

2. 修复方案 #

方案一：截断微秒部分适配MongoDB精度

直接修改SMT的格式配置，只解析前3位毫秒：

transforms.TimestampToIsoDate.format=yyyy-MM-dd'T'HH:mm:ss.SSS

此配置会忽略小数点后第4-5位，转换后得到正确的2023-08-22T13:52:50.134Z。

方案二：升级Confluent Connect版本（若可行）

Confluent Connect 7.4+版本优化了TimestampConverter的微秒解析逻辑，支持用SSSSSS格式正确解析微秒。升级后修改配置：

transforms.TimestampToIsoDate.format=yyyy-MM-dd'T'HH:mm:ss.SSSSSS

转换器会自动将5位微秒截断为3位毫秒，适配MongoDB的ISODate类型。

方案三：预处理消息（保留微秒精度）

若需要保留微秒信息（MongoDB 4.0+支持用Decimal128存储），可通过以下方式处理：

• 用KSQL预处理Kafka消息，将5位毫秒拆分为毫秒+微秒两个字段后再写入MongoDB
• 自定义SMT，正确解析5位微秒格式的时间字符串，转换为带微秒的Timestamp对象

3. 验证方法 #

• 使用kafka控制台工具查看转换后的消息结构，确认时间字段：

kafka-console-consumer --bootstrap-server <你的Kafka地址> --topic topic-with-sink --from-beginning --formatter kafka.connect.json.JsonConverter | jq '.LastUpdated'

• 开启Connector的debug日志，查看TimestampConverter的解析过程

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