Kafka Sink Connector能否仅对符合条件的消息应用值转换？

完全可以实现该需求，你可以直接使用Kafka Connect 2.6及以上版本内置的谓词匹配+条件转换能力，无需修改现有创建、更新类消息的处理逻辑，也无需引入第三方插件。

核心实现逻辑：定义匹配规则筛选出deleted标识为true的记录，仅对这类记录执行Drop$Value转换，将其变为消息体为null的墓碑记录，触发Elasticsearch sink connector的自动删除逻辑，其余记录直接走原有同步流程。

配置示例 #

直接在你的Elasticsearch sink connector配置中增加如下参数即可：

# 1. 定义谓词，用于匹配需要执行删除的记录
predicates=isDeletedRecord
predicates.isDeletedRecord.type=org.apache.kafka.connect.transforms.predicates.RecordValueMatches
# 可根据你的实际字段结构调整JSONPath表达式，嵌套字段可写成类似$.payload.deleted == true的格式
predicates.isDeletedRecord.predicate=$.deleted == true

# 2. 定义条件转换，仅对匹配谓词的记录丢弃消息体
transforms=dropDeletedValue
transforms.dropDeletedValue.type=org.apache.kafka.connect.transforms.Drop$Value
transforms.dropDeletedValue.predicate=isDeletedRecord
# 配置为false代表仅对匹配谓词的记录执行转换，不匹配的记录直接透传
transforms.dropDeletedValue.negate=false

注意事项 #

• 若你的Kafka Connect版本低于2.6，需要先升级版本，或者引入第三方转换插件实现类似的条件判断逻辑
• 注意配置的JSONPath表达式要和你的消息体结构完全匹配，避免匹配不到或者误匹配的情况
• 转换执行后符合条件的记录会变成标准墓碑记录，Elasticsearch sink connector会自动按主键删除对应ES文档，不需要修改其他同步配置

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