如何在Kafka JDBC Sink中移除消息里的字符？

要解决Kafka消息中包含空字符导致JDBC Sink写入数据库失败的问题，最直接的方式是利用Kafka Connect的RegexReplace转换，在消息到达Sink前自动移除空字符。以下是具体实现方案：

方案1：移除所有字段中的空字符 #

如果需要对消息中所有字符串字段的空字符进行清理，可针对整个消息值做全局替换。修改你的Sink配置，添加如下转换参数：

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "mode": "bulk",
    "table.name.format": "<Schema_name>.${topic}",
    "auto.evolve": "true",
    "tasks.max": "1",
    "topics": "<topic_name>",
    "name": "<topic_name>",
    "auto.create": "true",
    "connection.url": "jdbc:postgresql://<IP_address>:<port_number>/<db_name>?user=<user_name>&password=<password>",
    "insert.mode": "insert",
    // 新增转换配置
    "transforms": "removeAllNullChars",
    "transforms.removeAllNullChars.type": "org.apache.kafka.connect.transforms.RegexReplace$Value",
    "transforms.removeAllNullChars.regex": "\\u0000",
    "transforms.removeAllNullChars.replacement": ""
}

参数说明：

• transforms：定义转换任务名称，可自定义
• type：指定使用RegexReplace$Value，表示对整个消息值进行正则替换
• regex：匹配空字符\u0000（JSON配置中需转义反斜杠，故写为\\u0000）
• replacement：替换为空字符串，即移除匹配到的空字符

方案2：仅移除指定字段中的空字符 #

如果只需要清理特定字段（比如示例中的col2），可使用字段级别的转换：

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "mode": "bulk",
    "table.name.format": "<Schema_name>.${topic}",
    "auto.evolve": "true",
    "tasks.max": "1",
    "topics": "<topic_name>",
    "name": "<topic_name>",
    "auto.create": "true",
    "connection.url": "jdbc:postgresql://<IP_address>:<port_number>/<db_name>?user=<user_name>&password=<password>",
    "insert.mode": "insert",
    // 新增转换配置
    "transforms": "removeCol2NullChars",
    "transforms.removeCol2NullChars.type": "org.apache.kafka.connect.transforms.RegexReplace$Field",
    "transforms.removeCol2NullChars.field": "col2",
    "transforms.removeCol2NullChars.regex": "\\u0000",
    "transforms.removeCol2NullChars.replacement": ""
}

参数说明：

• type：使用RegexReplace$Field，表示针对指定字段处理
• field：指定要处理的字段名（如col2）

多字段处理 #

如果需要同时清理多个字段，可添加多个转换任务，用逗号分隔任务名称：

"transforms": "removeCol2NullChars,removeColXNullChars",
"transforms.removeCol2NullChars.type": "org.apache.kafka.connect.transforms.RegexReplace$Field",
"transforms.removeCol2NullChars.field": "col2",
"transforms.removeCol2NullChars.regex": "\\u0000",
"transforms.removeCol2NullChars.replacement": "",
"transforms.removeColXNullChars.type": "org.apache.kafka.connect.transforms.RegexReplace$Field",
"transforms.removeColXNullChars.field": "colX",
"transforms.removeColXNullChars.regex": "\\u0000",
"transforms.removeColXNullChars.replacement": ""

注意事项 #

• 确保Kafka Connect环境包含RegexReplace转换的依赖：Apache Kafka Connect和Confluent Platform默认已集成该基础转换，无需额外安装。
• 配置修改后，需重启Connector或通过REST API更新配置使其生效。

内容的提问来源于stack exchange，提问作者Dinesh Kumar L