Kafka Connect Avro枚举类型转字符串的技术问题咨询

我之前在使用Confluent Kafka Connect处理Avro枚举类型时也碰到过类似的转换问题，结合官方工具和自定义扩展，给你几个实用的解决方案：

方案1：使用内置的Cast单消息转换（SMT） #

这是最快捷的方式，不需要编写额外代码，利用Kafka Connect自带的Cast SMT就能完成枚举到字符串的转换。

你只需要在Kafka Connect的任务配置中添加以下transforms配置，指定要转换的枚举字段：

# 启用转换并命名
transforms=convertEnum
# 指定转换类型为值转换
transforms.convertEnum.type=org.apache.kafka.connect.transforms.Cast$Value
# 定义要转换的字段和目标类型：格式为「字段路径:string」
transforms.convertEnum.spec=your_enum_field_name:string

如果是嵌套字段的枚举，用点路径指定即可，比如user.status:string。

这个SMT会自动提取Avro枚举的字符串符号（也就是你在Schema中定义的枚举值），直接替换成字符串类型。

方案2：自定义单消息转换（SMT） #

如果内置的Cast SMT无法满足你的特殊需求（比如需要自定义枚举值的映射逻辑），可以自己写一个简单的SMT来处理：

示例代码（Java） #

import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import java.util.Map;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.ConfigDef.Importance;

public class EnumToStringTransform<R extends ConnectRecord<R>> implements Transformation<R> {
    private String targetField;

    @Override
    public void configure(Map<String, ?> configs) {
        this.targetField = (String) configs.get("target.field");
    }

    @Override
    public R apply(R record) {
        // 处理Struct类型的消息值（Avro解析后通常是Struct）
        if (record.value() instanceof Struct) {
            Struct struct = (Struct) record.value();
            Object enumValue = struct.get(targetField);
            if (enumValue instanceof Enum) {
                // 构建新的Schema，将目标字段改为字符串类型
                Schema newSchema = struct.schema().field(targetField).schema().type() == Schema.Type.ENUM
                        ? SchemaBuilder.struct(struct.schema())
                                .field(targetField, Schema.STRING_SCHEMA)
                                .build()
                        : struct.schema();
                
                // 创建新的Struct，替换枚举值为字符串
                Struct newStruct = new Struct(newSchema);
                for (String field : struct.schema().fields()) {
                    if (field.equals(targetField)) {
                        newStruct.put(field, ((Enum<?>) enumValue).name());
                    } else {
                        newStruct.put(field, struct.get(field));
                    }
                }
                return record.newRecord(record.topic(), record.kafkaPartition(), record.keySchema(), record.key(), newSchema, newStruct, record.timestamp());
            }
        }
        // 非Struct类型或无枚举字段时直接返回原记录
        return record;
    }

    @Override
    public void close() {}

    @Override
    public ConfigDef config() {
        return new ConfigDef()
                .define("target.field", Type.STRING, Importance.HIGH, "The name of the enum field to convert to string");
    }
}

使用步骤 #

1. 把代码打包成Jar文件
2. 将Jar放到Kafka Connect的插件目录（对应worker配置中的plugin.path）
3. 在Connect任务配置中添加transforms：

transforms=customEnumConvert
transforms.customEnumConvert.type=com.your.package.EnumToStringTransform
transforms.customEnumConvert.target.field=your_enum_field_name

方案3：验证Schema Registry中的枚举定义 #

有时候自动提交的Schema可能存在不符合预期的情况，你可以通过Schema Registry的API查看当前提交的Schema，确认枚举的定义是否正确：

curl -X GET http://your-schema-registry-host:8081/subjects/your_topic_name-value/versions/latest

如果Schema中的枚举符号和你应用中定义的不一致，可能需要调整应用端的Schema生成逻辑，或者手动提交正确的Schema到注册表（不过你提到是自动提交，所以优先检查应用端的Schema定义）。

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