如何在Kafka HTTP Sink Connector中使用Transform替换字段值

你的问题出在选错了Transform类型——ReplaceField是用来替换字段名称的，而你需要的是替换字段值，而且当前配置还错误地指向了$Key（处理消息Key），但你的目标是修改消息Value里的id字段值。

要实现按条件替换id字段值，你需要用RegexReplace Transform，它可以针对指定字段的内容做正则匹配替换。以下是修改后的完整连接器配置：

{
    "connector.class": "io.confluent.connect.http.HttpSinkConnector",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "topics": "product-topic",
    "tasks.max": "1",
    "http.api.url": "http://localhost:8080/product/create",
    "reporter.bootstrap.servers": "localhost:9092",
    "reporter.error.topic.name": "error-responses",
    "reporter.result.topic.name": "success-responses",
    "reporter.error.topic.replication.factor": "1",
    "confluent.topic.replication.factor": "1",
    "errors.tolerance": "all",
    "value.converter.schemas.enable": "false",
    "batch.json.as.array": "true",
    "name": "Product-Connector",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "reporter.result.topic.replication.factor": "1",
    "transforms": "ReplaceMobile,ReplaceComputer",
    "transforms.ReplaceMobile.type": "org.apache.kafka.connect.transforms.RegexReplace$Value",
    "transforms.ReplaceMobile.field": "id",
    "transforms.ReplaceMobile.regex": "test\\.product\\.mobile",
    "transforms.ReplaceMobile.replacement": "test.product.iPhone",
    "transforms.ReplaceComputer.type": "org.apache.kafka.connect.transforms.RegexReplace$Value",
    "transforms.ReplaceComputer.field": "id",
    "transforms.ReplaceComputer.regex": "test\\.product\\.computer",
    "transforms.ReplaceComputer.replacement": "test.product.mac"
}

配置说明 #

• 定义了两个Transform任务：ReplaceMobile和ReplaceComputer，分别处理两种id值的替换逻辑
• 每个Transform的类型为RegexReplace$Value，表示对消息的Value部分进行处理（而非Key）
• field参数指定要操作的目标字段是id
• regex是要匹配的原字段值：因为.在正则中是通配符，所以需要用\\转义，确保精确匹配完整字符串
• replacement是替换后的目标值

这样配置后，就能实现你预期的字段值替换效果，同时保留其他字段（如price）的内容不变。

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