使用EmbeddedKafka测试@KafkaListener时反序列化数据异常求助

核心问题现象 #

使用EmbeddedKafka测试@KafkaListener时，自定义KafkaDtoDeserializer接收到的byte[] data是消息主题（如topic-1）的字节数组，而非序列化后的KafkaDto数据；但连接真实Kafka时功能正常。

可能原因及对应解决步骤 #

1. 测试用Producer的序列化器逻辑错误 #

这是最可能的原因：你在测试中使用的kafkaDtoSerializer的serialize方法错误地将主题字符串作为序列化结果返回，而非KafkaDto对象的字节数组。

排查与修复：

检查kafkaDtoSerializer的实现，确保其正确序列化KafkaDto对象。例如正确的JSON序列化实现应该是：

@Component
@RequiredArgsConstructor
public class KafkaDtoSerializer implements Serializer<KafkaDto> {
    private final ObjectMapper mapper;

    @Override
    public byte[] serialize(String topic, KafkaDto data) {
        try {
            return mapper.writeValueAsBytes(data);
        } catch (JsonProcessingException e) {
            throw new SerializationException("Failed to serialize KafkaDto", e);
        }
    }
}

确认测试环境中注入的kafkaDtoSerializer是上述正确实现，而非返回主题字节的错误版本。

2. KafkaListener容器工厂未绑定自定义ConsumerFactory #

你的@KafkaListener指定了containerFactory = "kafkaListenerContainerFactory"，但如果该容器工厂Bean未显式绑定你定义的consumerFactoryDebezium，Spring会自动创建默认容器工厂，可能导致反序列化配置不生效（虽然这不是当前主题字节问题的直接原因，但会引发其他反序列化异常）。

修复：

显式创建kafkaListenerContainerFactoryBean并关联自定义ConsumerFactory：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, KafkaDto> kafkaListenerContainerFactory(
        ConsumerFactory<String, KafkaDto> consumerFactoryDebezium) {
    ConcurrentKafkaListenerContainerFactory<String, KafkaDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactoryDebezium);
    // 可添加其他配置如并发数、 Ack模式等
    return factory;
}

3. EmbeddedKafka的Bootstrap地址未正确覆盖配置 #

确保测试中Consumer的bootstrap.servers指向EmbeddedKafkaBroker的地址，而非真实Kafka。

修复：

在@EmbeddedKafka注解中显式指定bootstrapServersProperty：

@SpringBootTest
@EmbeddedKafka(topics = {"topic-1", "topic-2"}, bootstrapServersProperty = "spring.kafka.bootstrap-servers")
public class KafkaTest {...}

此配置会自动将Spring Boot的spring.kafka.bootstrap-servers覆盖为EmbeddedKafka的地址，确保Consumer连接到测试用Kafka集群。

4. ProducerRecord构造参数顺序错误 #

确认你创建ProducerRecord时参数顺序正确，避免将主题或其他值误作为消息value传入。你的代码中：

new ProducerRecord<>(kafkaConsumerProperties.getTopic(), "", kafkaDto);

是正确的（参数顺序为topic, key, value），但需再次确认kafkaConsumerProperties.getTopic()是你监听的目标主题，且kafkaDto是正确的消息体对象。

额外注意点 #

你的consumerFactoryDebezium配置存在重复定义：既在properties中设置了VALUE_DESERIALIZER_CLASS_CONFIG，又在DefaultKafkaConsumerFactory构造时传入了ErrorHandlingDeserializer。建议移除properties中的VALUE_DESERIALIZER_CLASS_CONFIG配置，以构造参数传入的deserializer为准：

@Bean
public ConsumerFactory<String, KafkaDto> consumerFactoryDebezium(KafkaConsumerProperties kafkaConsumerProperties) {
    Map<String, Object> properties = new HashMap<>(kafkaProperties.buildConsumerProperties());
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    // 移除此行重复配置
    // properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,KafkaDtoDeserializer.class);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroupId());
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProperties.isEnableAutoCommit());

    return new DefaultKafkaConsumerFactory<>(properties, new StringDeserializer(),
        new ErrorHandlingDeserializer<>(kafkaDtoDeserializer));
}

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