Debezium SQL Server源连接器DATE类型数据脱敏失效问题

核心问题在于你使用的column.mask.with.3.chars配置仅对字符串类型字段（如NVARCHAR/VARCHAR）生效，无法处理数值类型（BIGINT）和日期类型（DATE）字段，这就是c_name脱敏正常，而quantity和c_birthday未生效的原因。

解决步骤 #

1. 使用对应类型的专属脱敏配置 #

Debezium针对不同数据类型提供了专属掩码配置项：

• 数值类型（如BIGINT）：用column.mask.number，格式为数据库名.表名.列名:替换值
• 日期类型（如DATE）：用column.mask.date，格式为数据库名.表名.列名:替换日期
• 字符串类型：保留原column.mask.with.3.chars配置

2. 修改后的连接器配置示例 #

{"name": "customer-adhoc-5",
    "config": {
        "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
        "tasks.max" : "1",
        "topic.prefix" : "CDC",
        "database.hostname" : "sqlserver",
        "database.port" : "*****",
        "database.user" : "*****",
        "database.password" : "*****",
        "database.names" : "testDB",
        "schema.history.internal.kafka.bootstrap.servers" : "kafka:9092",
        "schema.history.internal.kafka.topic": "schema-changes.inventory",
        "include.schema.changes": "true",
        "snapshot.mode": "schema_only",
        "decimal.handling.mode": "double",
        "database.encrypt": "false",
        "table.include.list": "dbo.orders,dbo.staff",
        // 仅保留字符串类型字段
        "column.mask.with.3.chars": "testDB.dbo.staff.c_name",
        // BIGINT类型字段掩码：将quantity替换为0（可按需调整）
        "column.mask.number": "testDB.dbo.orders.quantity:0",
        // DATE类型字段掩码：将c_birthday替换为固定日期（可按需调整）
        "column.mask.date": "testDB.dbo.staff.c_birthday:1900-01-01",
        "schema.history.internal.store.only.captured.tables.ddl": "true",
        "schema.history.internal.store.only.captured.databases.ddl": "true",
        "incremental.snapshot.allow.schema.changes" : "true" ,
        "schema.name.adjustment.mode": "avro",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "****",
        "key.converter.schema.registry.url": "****"
    }
}

3. 灵活自定义脱敏规则（可选） #

如果需要更复杂的脱敏逻辑（比如数值保留前几位、日期只保留年份），可以使用Debezium的MaskField单消息转换（SMT），示例配置如下：

// 添加到config节点中
"transforms": "maskQuantity,maskBirthday",
"transforms.maskQuantity.type": "io.debezium.transforms.MaskField$Value",
"transforms.maskQuantity.fields": "quantity",
"transforms.maskQuantity.replacement": "0", // 可替换为自定义格式如"****"
"transforms.maskBirthday.type": "io.debezium.transforms.MaskField$Value",
"transforms.maskBirthday.fields": "c_birthday",
"transforms.maskBirthday.replacement": "1900-01-01" // 或自定义日期格式

注意事项 #

• 修改配置后必须重启连接器，确保新规则生效
• 若使用增量快照模式，需确认新的数据库变更会应用新的脱敏规则
• 检查Schema Registry中的Schema是否更新，确保字段配置与实际规则匹配

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