我太懂你这种两难的处境了——MySQL那边把空字符串塞进NOT NULL字段绕开约束，结果到Oracle这儿直接触发NULL违反约束，Sink连接器崩溃；要是逐个字段配SMT替换，又完全失去了Debezium自动化同步的意义。给你几个全局处理的可行方案，不用硬编码字段名：

方案一：用Kafka Connect的ScriptTransform全局遍历处理空字符串 #

这是最灵活的全局解决方案，通过Groovy脚本遍历消息里的所有字符串字段，把空字符串替换成Oracle不会识别为NULL的值（比如空格）。

步骤1：修改Sink连接器配置 #

在你现有的JDBC Sink配置里添加ScriptTransform相关参数：

{
  "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
  "tasks.max": "1",
  "connection.url": "jdbc:oracle:thin:@ldap://ldpap.com:3060/testdb,cn=OracleContext,dc=domain,dc=priv",
  "connection.username": "username",
  "connection.password": "password",
  "insert.mode": "insert",
  "schema.evolution": "basic",
  "database.time_zone": "UTC",
  "topics.regex": "(DBNAME\\.DBNAME\\.([^.]+)$)",
  "quote.identifiers": "true",
  // 添加以下Transform配置
  "transforms": "fixEmptyStrings",
  "transforms.fixEmptyStrings.type": "org.apache.kafka.connect.transforms.Script",
  "transforms.fixEmptyStrings.script": "global_empty_string_fix.groovy"
}

步骤2：编写Groovy处理脚本 #

创建global_empty_string_fix.groovy文件，内容如下：

import org.apache.kafka.connect.data.Schema
import org.apache.kafka.connect.data.Struct

// 定义处理单个Struct的逻辑
def processStruct(Struct struct) {
    struct.schema().fields().each { field ->
        def value = struct.get(field)
        // 只处理字符串类型且值为空字符串的字段
        if (field.schema().type() == Schema.Type.STRING && value == "") {
            // 替换为空格（Oracle不会将空格视为NULL），也可换成业务允许的其他默认值
            struct.put(field, " ")
        }
    }
    return struct
}

// 处理Debezium的CDC消息结构（包含before/after）
if (value.schema().type() == Schema.Type.STRUCT) {
    def payload = value.get("payload")
    if (payload != null) {
        // 处理after部分（新增/更新后的数据）
        if (payload.get("after") != null) {
            payload.put("after", processStruct(payload.get("after")))
        }
        // 处理before部分（更新前的数据，upsert场景可能需要）
        if (payload.get("before") != null) {
            payload.put("before", processStruct(payload.get("before")))
        }
    }
}
return value

这个脚本会自动遍历消息中before和after里的所有字段，只要是字符串类型且值为空，就替换成空格，完美绕开Oracle的空字符串=NULL特性。

方案二：在Source端提前处理（可选） #

如果你希望整个Kafka集群里的消息都是处理后的数据，也可以把同样的ScriptTransform配置在Debezium MySQL Source Connector里，这样Sink端直接消费就行，不用额外配置。配置逻辑和上面一致，只是把Transform加到Source的配置中。

关键注意事项 #

• 业务适配：如果空格不符合你的业务需求，可以把脚本里的空格换成业务允许的默认值（比如"N/A"、"0"等，根据字段类型调整）。
• 依赖检查：确保Kafka Connect的类路径里有Groovy的依赖（大部分官方Debezium镜像已经包含，如果没有，需要添加groovy-all.jar）。
• 测试验证：先拿单个测试表验证脚本效果，确认没问题后再全局启用。

另外提一句：你说的换成insert.mode: upsert其实解决不了根本问题——Oracle还是会把空字符串识别为NULL，只要字段有NOT NULL约束，照样会触发错误，所以数据层面的处理才是核心。

备注：内容来源于stack exchange，提问作者grizzo