Spring Boot下application.properties配置自定义Kafka Avro反序列化器问题

首先明确结论：Spring Boot 默认的application.properties配置方式不支持直接调用带参数的反序列化器构造方法，Kafka的反序列化器初始化逻辑要求配置指定的类必须存在公共无参构造方法，否则就会抛出你遇到的异常。

你可以选择两种方案解决： #

方案一：改造自定义反序列化器，通过配置实现全参数化（无需额外代码配置） #

Kafka的Deserializer接口自带configure方法，会在反序列化器初始化时被调用，所有spring.kafka.consumer.properties开头的配置都会传入该方法的参数中，你可以改造你的反序列化器适配无参构造：

1. 给AvroDeserializer添加公共无参构造方法
2. 将Schema加载逻辑迁移到configure方法中，从传入的配置参数里读取本地Schema路径进行加载

示例代码：

public class AvroDeserializer<T> implements Deserializer<T> {
    private Schema schema;
    // 新增公共无参构造
    public AvroDeserializer() {}

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 读取自定义配置的Schema路径
        String schemaPath = (String) configs.get("custom.avro.schema.path");
        try (InputStream is = getClass().getResourceAsStream(schemaPath)) {
            this.schema = new Schema.Parser().parse(is);
        } catch (IOException e) {
            throw new RuntimeException("Avro Schema加载失败", e);
        }
    }

    @Override
    public T deserialize(String topic, byte[] data) {
        // 原有反序列化逻辑，直接使用上面加载的schema即可
    }

    @Override
    public void close() {}
}

对应的application.properties配置：

# 原有反序列化器配置
spring.kafka.consumer.value-deserializer=com.example.AvroDeserializer
# 自定义Schema路径配置，会自动传入configure方法
spring.kafka.consumer.properties.custom.avro.schema.path=/avro/your-schema.avsc
# 其他Kafka消费者配置正常添加即可
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=your-group

方案二：通过代码手动配置消费者工厂（无需修改原有反序列化器） #

如果你不想改动已经写好的带参构造反序列化器，可以直接手动注册ConsumerFactory和Kafka监听容器工厂的Bean，手动实例化反序列化器传入Schema参数：

示例代码：

@Configuration
public class KafkaConsumerConfig {
    @Value("${custom.avro.schema.path}")
    private String schemaPath;

    @Bean
    public ConsumerFactory<String, Object> avroConsumerFactory() throws IOException {
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group");
        // 手动加载Schema
        Schema schema = new Schema.Parser().parse(getClass().getResourceAsStream(schemaPath));
        // 直接实例化你的带参构造反序列化器
        AvroDeserializer<Object> valueDeserializer = new AvroDeserializer<>(schema);
        return new DefaultKafkaConsumerFactory<>(consumerProps, new StringDeserializer(), valueDeserializer);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> avroKafkaListenerContainerFactory() throws IOException {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(avroConsumerFactory());
        return factory;
    }
}

使用时只需要在@KafkaListener注解中指定该工厂即可：

@KafkaListener(topics = "your-topic", containerFactory = "avroKafkaListenerContainerFactory")
public void listen(YourAvroModel message) {
    // 处理消息逻辑
}

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