Kafka Connect使用Schema Registry时自定义SubjectNameStrategy的configurable方法未调用

• 自定义MySubjectNameStrategy类未正确实现org.apache.kafka.common.Configurable接口，只有实现该接口的类，才会被Kafka Connect的配置机制自动调用configure方法
• 使用的Confluent Avro Converter版本低于6.2.0，该版本之前的AvroConverter在初始化SubjectNameStrategy实例后，不会主动触发configure方法调用，属于已知版本缺陷
• 自定义策略的配置项前缀错误，导致参数无法被传递到configure方法的入参Map中

步骤1：补全类实现要求 #

确保你的MySubjectNameStrategy同时满足两个实现要求：

• 实现io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy接口
• 实现org.apache.kafka.common.Configurable接口，重写configure(Map<String, ?> configs)方法
  示例代码片段：

package com.test;

import io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy;
import org.apache.kafka.common.Configurable;
import java.util.Map;

public class MySubjectNameStrategy implements SubjectNameStrategy, Configurable {
    private String customParam;

    @Override
    public void configure(Map<String, ?> configs) {
        // 此处可获取所有converter相关的配置参数
        this.customParam = (String) configs.get("value.converter.custom.param");
    }

    // 原有subjectName相关实现逻辑保持不变
}

步骤2：版本适配处理 #

如果你的Confluent平台版本低于6.2.0，有两个可选方案：

• 方案1：升级kafka-connect-avro-converter依赖到6.2.0及以上版本，注意和Schema Registry版本保持一致
• 方案2：如果无法升级版本，可以在subjectName方法第一次被调用时手动加载配置参数，仅作为临时兼容方案

步骤3：配置参数传入规则 #

给自定义策略传递专有配置时，配置项需要带上对应converter的前缀：

• 如果使用value.converter的自定义策略，参数格式为value.converter.你的参数名
• 如果使用key.converter的自定义策略，参数格式为key.converter.你的参数名
  只有符合前缀规则的配置，才会被包含到configure方法的入参configs中，示例配置：

# 原有配置保持不变
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=https://schema-registry-host:8082
value.converter.auto.register.schemas=false
value.converter.schema.registry.ssl.truststore.location=/path
value.converter.use.latest.version=true
value.converter.latest.compatibility.strict=false
value.converter.key.subject.name.strategy=com.test.MySubjectNameStrategy
value.converter.value.subject.name.strategy=com.test.MySubjectNameStrategy
# 新增自定义参数，前缀和converter前缀保持一致
value.converter.custom.param=testValue

验证 #

修改完成后重启Kafka Connect Worker，触发一次数据同步即可验证configure方法已被正常调用、配置参数可正常获取。

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