Kafka Sink Connector无法对同字段两次应用TimestampConverter转换求助

问题 #

Debezium Source Connector向Kafka生成的created_on字段存在两种时间戳格式：

1. "2023-02-26T18:30:00Z"
2. "2022-10-27T09:21:45.522779Z"

我尝试在Kafka Sink Connector中对该字段两次应用TimestampConverter转换来适配两种格式，但只有第一个转换生效，第二个未生效。当前转换配置如下：

"transforms.tscreatedon.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.tscreatedon.field": "created_on",
"transforms.tscreatedon.format": "yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'",
"transforms.tscreatedon.target.type": "Timestamp",
"transforms.tscreatedon2.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.tscreatedon2.field": "created_on",
"transforms.tscreatedon2.format": "yyyy-MM-dd'T'HH:mm:ss'Z'",
"transforms.tscreatedon2.target.type": "Timestamp"

请问配置哪里出错了？如何实现适配两种格式的需求？

分析与解决方法 #

问题根源 #

Kafka Connect的转换是顺序执行的，且TimestampConverter转换成功后会把原字符串字段替换成Timestamp类型：

• 当第一个转换处理完带微秒的时间戳后，字段类型已变为Timestamp，第二个转换再处理时，因目标字段不是字符串，会直接跳过，所以不生效。
• 如果遇到不带微秒的时间戳，第一个转换会因格式不匹配直接失败，根本到不了第二个转换步骤。

解决方案 #

不能通过两次TimestampConverter处理两种格式，推荐以下思路：

• 自定义SMT转换：编写自定义的单消息转换（SMT）逻辑，在转换时依次尝试两种格式解析，直到成功将字符串转为Timestamp类型。
• 统一源端输出格式：修改Debezium Source Connector的配置，让源端直接输出统一格式的时间戳（比如统一输出带微秒的字符串，或直接输出Timestamp类型），从根源避免格式不一致问题。
• 用KSQL预处理：在Sink Connector之前，通过KSQL创建流，对created_on字段做条件解析——先尝试解析带微秒的格式，失败则解析不带微秒的格式，生成统一的Timestamp字段后再给Sink Connector消费。

临时折中方案（不推荐长期使用） #

如果不想编写自定义逻辑或使用KSQL，可以尝试调整转换顺序并开启容错：

1. 把不带微秒格式的转换放在前面，带微秒的放在后面。
2. 给两个转换都添加ignore.errors=true参数，让转换失败时跳过，继续执行下一个。

示例配置片段：

"transforms.tscreatedon.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.tscreatedon.field": "created_on",
"transforms.tscreatedon.format": "yyyy-MM-dd'T'HH:mm:ss'Z'",
"transforms.tscreatedon.target.type": "Timestamp",
"transforms.tscreatedon.ignore.errors": "true",
"transforms.tscreatedon2.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.tscreatedon2.field": "created_on",
"transforms.tscreatedon2.format": "yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'",
"transforms.tscreatedon2.target.type": "Timestamp",
"transforms.tscreatedon2.ignore.errors": "true"

注意：这种方法可能导致部分消息无法被正确转换，仅作为临时过渡方案。

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