Debezium捕获SQL Server自定义类型字段丢失及警告问题求助

问题背景 #

• 环境：Debezium 3.0.2 Final、SQL Server 2017
• 核心现象：捕获T1表变更事件时，每次修改行数据都会触发警告：

  WARN Column 'myCol1' available in capture table not found among source table columns (io.debezium.connector.sqlserver.SqlServerChangeTablePointer)

• 衍生问题：CString:varchar(30)等自定义数据类型的字段数据无法写入Kafka主题
• 已尝试方案：自定义类型转换器、配置column.include.list，但问题未解决

自定义转换器代码：

public class CustomSqlTypeConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
    private static final Logger LOGGER = LoggerFactory.getLogger(CustomSqlTypeConverter.class);

    @Override
    public void configure(Properties props) {

    }

    @Override
    public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {
        String columnType = column.typeName().toUpperCase();
        switch (columnType) {
            case "NSTRING":
            case "CSTRING":
            case "QSTRINGDT":
            case "TCODE":
            case "TINFO":
            case "TINFO_U":
                registration.register(SchemaBuilder.string().optional(), rs -> rs.toString());
                LOGGER.info("Mapped SQL type CONTRACTNAME to String.");
                break;

            default:
                LOGGER.debug("No custom mapping for type: {}", columnType);
        }
    }
}

连接器配置片段：

"column.include.list": "dbo.table1.(pkCol|myCol1|myCol2)"

问题原因 #

1. 自定义类型元数据不匹配：SQL Server变更捕获表中，自定义类型会以基础数据类型（如varchar(30)）存储，但Debezium在对比源表与变更表的列时，会校验类型名称，导致自定义类型列无法被识别匹配，触发警告并跳过同步。
2. 自定义转换器未生效：转换器代码中的日志信息与实际处理类型不匹配（日志写的是CONTRACTNAME，但实际处理的是CString等类型），无法确认是否触发了映射逻辑；同时可能存在转换器未被连接器正确加载的情况。
3. 列匹配规则问题：column.include.list使用正则表达式可能存在匹配偏差，导致自定义类型列未被正确纳入同步范围。

解决方案 #

1. 修复自定义转换器并确保加载 #

• 修正转换器日志：将日志语句改为LOGGER.info("Mapped SQL type {} to String.", columnType);，方便排查是否触发了自定义类型映射。
• 完善连接器配置：添加转换器注册参数，确保连接器加载自定义转换器：

  "converters": "customSqlConverter",
  "customSqlConverter.type": "com.your.package.CustomSqlTypeConverter",
  "customSqlConverter.schema.name": "custom.sql.types"
  

  注意替换com.your.package为实际的转换器包路径。

2. 调整列包含列表配置 #

避免使用正则表达式，直接指定列名列表，确保大小写与源表完全一致：

"column.include.list": "dbo.T1.pkCol,dbo.T1.myCol1,dbo.T1.myCol2"

3. 重置SQL Server变更捕获 #

若变更捕获表元数据与源表不一致，可重置变更捕获同步元数据：

-- 禁用目标表的变更捕获
ALTER TABLE dbo.T1 DISABLE CHANGE_TRACKING;
-- 重新启用变更捕获，确保元数据同步
ALTER TABLE dbo.T1 ENABLE CHANGE_TRACKING WITH (TRACK_COLUMNS_UPDATED = ON);

4. 字段转换适配变更表结构 #

如果自定义类型在变更表中以基础类型存储，可通过字段转换匹配列：

"column.transforms": "unwrap,mapMyCol1",
"column.transforms.mapMyCol1.type": "org.apache.kafka.connect.transforms.RenameField$Value",
"column.transforms.mapMyCol1.field": "myCol1",
"column.transforms.mapMyCol1.target": "myCol1"

（注：若变更表中列名与源表有差异，需调整target值为变更表中的实际列名）

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