如何在Kafka Connect自定义转换器中传入并读取环境变量

嘿，这个场景我刚好在项目里实践过，咱们一步步来解决你的问题：

1. 能不能在configure阶段读取环境变量？ #

完全可以！而且这是推荐的时机。configure(Map<String, ?> configs)是Kafka Connect组件（包括转换器）的初始化方法，在组件启动时只会调用一次，在这里读取环境变量或配置参数，既能避免重复读取的性能损耗，也符合组件的生命周期规范。

2. 访问环境变量的两种方式 #

方式一：直接在代码中读取System.getenv() #

最简单的方式就是在转换器代码里直接调用System.getenv("CLUSTER")，不管是在configure方法还是转换方法里都能生效。比如：

public class CustomJsonConverter extends JsonConverter {
    private String clusterName;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        super.configure(configs, isKey);
        // 直接读取环境变量
        this.clusterName = System.getenv("CLUSTER");
        // 做个非空校验，避免后续逻辑报错
        if (clusterName == null || clusterName.isBlank()) {
            throw new IllegalArgumentException("环境变量CLUSTER未设置或为空");
        }
    }

    // 在消息转换时追加集群名称
    @Override
    public SchemaAndValue convertToConnectData(String topic, byte[] value) {
        SchemaAndValue original = super.convertToConnectData(topic, value);
        // 假设原始值是Struct类型，追加cluster字段
        if (original.value() instanceof Struct) {
            Struct originalStruct = (Struct) original.value();
            // 扩展Schema，添加cluster字段
            Schema newSchema = originalStruct.schema().builder()
                    .field("cluster", Schema.STRING_SCHEMA)
                    .build();
            // 构建新的Struct并填充原有字段+cluster
            Struct newStruct = new Struct(newSchema);
            originalStruct.schema().fields().forEach(field -> {
                newStruct.put(field.name(), originalStruct.get(field));
            });
            newStruct.put("cluster", clusterName);
            return new SchemaAndValue(newSchema, newStruct);
        }
        // 如果是Map类型，直接put即可
        if (original.value() instanceof Map) {
            Map<String, Object> map = new HashMap<>((Map<String, Object>) original.value());
            map.put("cluster", clusterName);
            return new SchemaAndValue(original.schema(), map);
        }
        return original;
    }
}

这种方式的优点是简单直接，不需要额外配置；缺点是灵活性差——如果后续需要给不同的转换器实例设置不同的集群名称，就没法通过配置区分了。

方式二：通过configs映射传入（推荐） #

这种方式更符合Kafka Connect的配置设计理念，把环境变量的值注入到configs中，再在configure方法里读取。具体步骤如下：

步骤1：配置Kafka Connect支持环境变量替换

首先需要在Connect的全局配置文件（connect-standalone.properties或connect-distributed.properties）中添加配置提供者，让Connect能够解析配置中的环境变量占位符：

# 启用环境变量配置提供者
config.providers=env
config.providers.env.class=org.apache.kafka.common.config.provider.EnvironmentVariableConfigProvider

步骤2：在转换器配置中引用环境变量

接下来，在你的连接器配置文件（或分布式模式下的REST请求体）中，给自定义转换器添加一个配置项，值用${CLUSTER}引用环境变量：
比如在连接器配置里：

# 指定自定义转换器
value.converter=com.yourcompany.CustomJsonConverter
# 给转换器传入集群名称，从环境变量CLUSTER读取
value.converter.cluster=${CLUSTER}

如果是分布式模式，通过REST API提交配置时，JSON体可以这么写：

{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "value.converter": "com.yourcompany.CustomJsonConverter",
    "value.converter.cluster": "${CLUSTER}",
    // 其他连接器配置...
  }
}

步骤3：在转换器的configure方法中读取配置

然后在自定义转换器的configure方法里，从configs中获取这个配置项：

public class CustomJsonConverter extends JsonConverter {
    private String clusterName;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        super.configure(configs, isKey);
        // 从configs中读取配置项，这里的key要和配置文件里的一致
        this.clusterName = (String) configs.get("cluster");
        // 非空校验
        if (clusterName == null || clusterName.isBlank()) {
            throw new IllegalArgumentException("转换器配置项cluster未设置或为空");
        }
    }

    // 消息转换逻辑和之前一样，使用clusterName追加字段
}

这种方式的优点是灵活可控，可以给不同的转换器实例设置不同的值，也方便通过Connect的配置管理功能进行调试和修改；缺点是需要多几步配置，但这是符合Kafka Connect最佳实践的方式。

3. 注意事项 #

• 环境变量大小写敏感：在Linux/Unix系统中，环境变量名称是大小写敏感的，确保代码里的CLUSTER和实际设置的环境变量名称一致。
• 分布式模式下的一致性：如果是分布式部署的Connect集群，所有Worker节点的CLUSTER环境变量值要保持一致，或者使用统一的配置中心来管理。
• 非空校验：一定要在configure阶段做非空校验，避免后续转换消息时出现空指针异常。

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