如何通过Debezium连接器将MySQL列注释作为元数据加入Avro Schema？

针对你的需求（通过Debezium MySQL连接器同步带特定注释的敏感列，并在Avro Schema中添加MY_CUSTOM_ATTRIBUTE自定义属性），下面分步骤解决你遇到的两个问题：

问题1：如何在自定义转换中获取MySQL列注释 #

Debezium 2.0开启schema.doc.from.column.comments=true后，会把MySQL列的注释同步到Kafka Connect字段Schema的两个位置：

• 字段Schema的doc属性直接等于列注释
• 字段Schema的parameters中会保留comment参数（值为列注释）

在自定义转换中，你可以通过Schema.Field的schema().doc()或者schema().parameters().get("comment")直接获取到列注释。

问题2：如何给Schema添加自定义属性 #

Kafka Connect的SchemaBuilder其实支持通过parameter(String key, String value)方法添加自定义参数，这些参数会被Confluent的AvroConverter自动映射为Avro Schema的自定义元数据属性，完全符合Avro规范。

完整实现步骤 #

1. 配置Debezium连接器，开启列注释同步

在你的Debezium MySQL连接器配置中添加：

schema.doc.from.column.comments=true

2. 编写自定义转换类

实现Kafka Connect的Transformation接口，遍历字段并给符合条件的列添加自定义属性（支持嵌套结构体处理）：

import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.transforms.Transformation;
import java.util.Map;

public class AddSensitiveAttribute implements Transformation<ConnectRecord> {
    private String sensitiveComment = "sensitive column";
    private static final String CUSTOM_ATTR = "MY_CUSTOM_ATTRIBUTE";

    @Override
    public void configure(Map<String, ?> configs) {
        // 支持通过配置自定义敏感注释内容
        if (configs.containsKey("sensitive.comment")) {
            this.sensitiveComment = configs.get("sensitive.comment").toString();
        }
    }

    @Override
    public ConnectRecord apply(ConnectRecord record) {
        if (record.valueSchema() == null) {
            return record;
        }

        // 递归处理Schema（包括嵌套结构体）
        Schema newSchema = processSchema(record.valueSchema());
        return record.newRecord(
                record.topic(), record.kafkaPartition(),
                record.keySchema(), record.key(),
                newSchema, record.value(),
                record.timestamp()
        );
    }

    // 递归处理所有字段（支持嵌套结构体）
    private Schema processSchema(Schema schema) {
        if (schema.type() != Schema.Type.STRUCT) {
            return schema;
        }

        SchemaBuilder builder = SchemaBuilder.copy(schema);
        // 复制原Schema的所有参数
        schema.parameters().forEach(builder::parameter);

        for (Schema.Field field : schema.fields()) {
            Schema fieldSchema = processSchema(field.schema());
            SchemaBuilder fieldBuilder = SchemaBuilder.copy(fieldSchema);

            // 获取列注释并判断是否为敏感列
            String comment = fieldSchema.doc();
            if (sensitiveComment.equals(comment)) {
                fieldBuilder.parameter(CUSTOM_ATTR, "true");
            }

            // 复制原字段的所有参数
            fieldSchema.parameters().forEach(fieldBuilder::parameter);
            builder.field(field.name(), fieldBuilder.build());
        }

        return builder.build();
    }

    @Override
    public void close() {}
}

3. 打包并部署自定义转换

把代码打包成Jar文件，放到Kafka Connect节点的plugin.path指定目录下，重启Connect服务生效。

4. 在连接器配置中启用自定义转换

# 启用转换
transforms=addSensitive
# 指定转换类的全限定名
transforms.addSensitive.type=com.yourcompany.transforms.AddSensitiveAttribute
# 可选：自定义敏感列的注释内容（如果你的注释不是"sensitive column"）
# transforms.addSensitive.sensitive.comment=你的敏感列注释

验证效果 #

同步完成后，在Schema Registry中查看对应主题的Avro Schema，敏感列的字段定义会包含MY_CUSTOM_ATTRIBUTE: "true"的自定义属性，示例如下：

{
  "name": "name",
  "type": "string",
  "MY_CUSTOM_ATTRIBUTE": "true"
}

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