Kafka Connect MaskField SMT 在墓碑上失败

Question

我想在特定主题的字段上应用 MaskField SMT。遇到墓碑事件（空值）时，MaskField SMT 失败并显示 Only Map objects supported in absence of schema for [mask fields], found: null。为了仅对选定的主题应用屏蔽，我使用了TopicNameMatches 谓词，但我没有找到一种方法来应用否定的RecordIsTombstone 以及。我发现对于某些 SMT，有一个选项可以让墓碑通过而不受转换的影响，但对于 MaskField 似乎无能为力。有没有办法在不编写自定义 SMT（使用 Kafka Connect 6.0.1）的情况下实现这一点？

我目前的配置：

transforms: mask_trf
transforms.mask_trf.type: org.apache.kafka.connect.transforms.MaskField$Value
transforms.mask_trf.fields: foo
transforms.mask_trf.replacement: ***
transforms.mask_trf.predicate: pred
predicates: pred
predicates.pred.type: org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.pred.pattern: t1

Answer 1

我最终创建了一个扩展 MaskField 的自定义 SMT。唯一的变化是在 apply() 中它检查它是否是墓碑，如果是，则返回未更改的记录。否则调用 MaskField 的 apply。