Confluent Elasticsearch Connector空值字段不覆盖已有字段如何实现

问题结论 #

当前Confluent Elasticsearch Sink Connector没有原生提供类似Datastax C* Connector nullToUnset的单字段null值忽略更新配置，behavior.on.null.values确实仅用于处理整条记录为null时的删除逻辑，不适用于单字段null的场景。

可行解决方案 #

推荐按优先级选择以下方案：

方案1：使用Kafka Connect内置DropNullFields SMT预处理（首选） #

这是成本最低、兼容性最好的方案，只需要修改Connector配置即可实现需求：

• 核心逻辑：在消息发送到ES Sink Connector之前，通过单消息转换（SMT）移除掉记录中所有值为null的字段，后续执行upsert操作时，只会更新请求中存在的非null字段，已有的字段值不会被null覆盖。
• 配置示例：

# 原有ES Connector配置保持不变，新增以下SMT配置
transforms=dropNullFields
transforms.dropNullFields.type=org.apache.kafka.connect.transforms.DropNullFields$Value
# 如有特殊字段即使为null也需要更新到ES，可配置排除列表，示例如下：
# transforms.dropNullFields.exclude.list=is_deleted,expire_time

• 效果验证：你示例中的后续 enriched 记录{id: 1, name: null, age: 34}经过SMT处理后会变为{id: 1, age: 34}，upsert后ES中最终结果就是预期的{id: 1, name: "Bob", age: 34}。

方案2：配置ES自定义更新脚本（适合字段固定场景） #

如果你的业务存在部分null字段需要更新、部分不需要的复杂规则，可通过ES Connector的update.script配置自定义更新逻辑：

write.method=upsert
update.script=for (entry in params.event.entrySet()) {if (entry.getValue() != null) {ctx._source[entry.getKey()] = entry.getValue()}}

该方案无需修改上游业务代码，但如果字段是动态新增的，需要同步维护脚本逻辑，维护成本高于SMT方案。

方案3：自定义修改ES Sink Connector源码（不推荐） #

如果前两种方案都无法满足需求，可以修改Connector构建更新请求的逻辑，过滤掉null字段后再发送到ES，该方案需要自行维护后续Connector版本升级，成本最高，非必要不选择。

注意事项 #

使用DropNullFields SMT时需要确保作为文档主键的id字段为必填字段，不会出现null值被过滤导致的主键缺失问题。

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