使用Kafka Connect时如何按Avro Schema批量转换所有时间戳字段

原生Kafka Connect提供的TimestampConverter单消息转换（SMT）本身不支持按Schema元数据批量匹配字段，只能硬编码指定单个字段名，你之前的配置里重复定义transforms.tsFormat.field本身就是无效写法——Properties配置里同key的后写入值会覆盖前值，最后实际只有最后一个配置的字段会被转换。
要实现自动识别所有io.debezium.time.Timestamp类型字段统一转换、新增字段无需改配置的需求，按优先级推荐以下方案：

不需要额外配置任何SMT，直接修改Debezium连接器的时间精度相关配置即可全局生效，所有被Debezium识别为时间戳的字段都会自动按规则输出，新增字段完全不需要调整配置：

# Debezium 1.9.0+ 版本支持，直接将所有时间类型字段输出为标准ISO 8601格式字符串
time.precision.mode = iso_string

如果你的Debezium版本在1.8及以下，无法升级，可以用以下配置输出Kafka Connect原生的Timestamp逻辑类型：

time.precision.mode = connect

这个模式下输出的字段自带org.apache.kafka.connect.data.Timestamp的Schema标记，下游组件（比如Elasticsearch、JDBC sink连接器）可以自动识别为时间类型，不需要额外转换。

注意点：

• 该配置对所有Debezium时间类型生效（Timestamp、Date、Time、MicroTimestamp等），不需要单独指定字段
• 配置生效后，所有新增的时间类型字段会自动遵循转换规则，不需要修改连接器配置重启任务

如果不想修改全局时间输出规则，或者Debezium版本太老无法升级，可以部署支持按Schema元数据匹配字段的通用SMT，一次部署后所有连接器可复用：
你可以直接使用社区维护的通用SMT包，也可以自己实现一个轻量SMT，核心逻辑非常简单：

1. 实现Kafka Connect的Transformation接口
2. 遍历消息Struct中的每一个字段，读取字段对应的Schema元数据
3. 判断Schema的name属性是否为io.debezium.time.Timestamp，对匹配的字段执行long到时间格式的转换
4. SMT配置时只需要指定目标格式（比如字符串、Timestamp类型），不需要传入任何字段列表
   这类SMT的代码量通常在50行以内，编译打包后放到Kafka Connect的插件目录即可使用，后续所有需要做同类时间转换的连接器都可以直接引用，不需要重复开发。

如果暂时无法修改连接器侧的配置，可以在消费端做统一处理：消费消息时先读取Avro Schema，遍历所有字段判断connect.name属性，对匹配到的时间字段做值转换。这个方案的缺点是每个消费端都要实现一遍转换逻辑，维护成本高，只适合临时应急使用。

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