Spring Boot单元测试中Kafka ConsumerRecord返回null问题排查

我来帮你排查下这个问题——既然kafka控制台能正常收到消息，说明生产者的发送逻辑是没问题的，问题大概率出在测试代码的消费配置或者测试方式上，给你几个可行的调试方向：

1. 检查消费者的位移配置 #

如果测试用的是固定的group.id，之前的消费位移可能导致新消息被直接跳过。你可以手动重置消费位移，或者在测试配置里指定auto.offset.reset=earliest，确保消费者从最开始的位置读取消息：

@Autowired
private ConsumerFactory<String, YourPOJO> consumerFactory;

@Before
public void setupConsumer() {
    KafkaConsumer<String, YourPOJO> consumer = (KafkaConsumer<String, YourPOJO>) consumerFactory.createConsumer("test-group");
    consumer.subscribe(Collections.singletonList("your-topic-name"));
    // 强制重置到最早位移
    consumer.seekToBeginning(consumer.assignment());
}

2. 用@EmbeddedKafka搭建独立测试环境 #

推荐配合@EmbeddedKafka注解来创建内嵌的Kafka集群，完全隔离外部环境，避免各种配置干扰：

@RunWith(SpringRunner.class)
@SpringBootTest
@DisplayName("Testing GlobalMessageTest")
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
public class NumberPlateSenderTest {

    @Autowired
    private KafkaTemplate<String, YourPOJO> kafkaTemplate;

    @Autowired
    private ConsumerFactory<String, YourPOJO> consumerFactory;

    @Test
    public void testSendAndReceive() throws InterruptedException, ExecutionException {
        // 发送测试消息
        YourPOJO testData = new YourPOJO();
        kafkaTemplate.send("your-topic", testData).get();

        // 创建消费者并等待消息
        KafkaConsumer<String, YourPOJO> consumer = (KafkaConsumer<String, YourPOJO>) consumerFactory.createConsumer("test-group");
        consumer.subscribe(Collections.singletonList("your-topic"));
        ConsumerRecords<String, YourPOJO> records = consumer.poll(Duration.ofSeconds(3));
        
        // 断言消息存在
        assertFalse(records.isEmpty());
        ConsumerRecord<String, YourPOJO> targetRecord = records.iterator().next();
        assertNotNull(targetRecord);
    }
}

3. 解决异步消费的同步问题 #

如果你的消费逻辑是基于@KafkaListener的异步监听，测试方法可能会因为执行太快，在消息被消费前就结束了。可以用CountDownLatch来做同步控制：

@Component
public class YourKafkaListener {
    private CountDownLatch latch;
    private ConsumerRecord<String, YourPOJO> receivedRecord;

    @KafkaListener(topics = "your-topic")
    public void listen(ConsumerRecord<String, YourPOJO> record) {
        this.receivedRecord = record;
        if (latch != null) {
            latch.countDown();
        }
    }

    // getter和setter方法
}

// 测试类里的方法
@Test
public void testWithSyncLatch() throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);
    yourKafkaListener.setLatch(latch);

    // 发送消息
    kafkaTemplate.send("your-topic", testData).get();
    // 等待最多5秒，直到收到消息
    assertTrue(latch.await(5, TimeUnit.SECONDS));
    assertNotNull(yourKafkaListener.getReceivedRecord());
}

4. 验证序列化配置一致性 #

虽然你说POJO能正常工作，但还是要确认测试环境的消费端反序列化配置和生产端完全匹配，比如在application-test.properties里：

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=com.yourpackage.YourCustomDeserializer
spring.kafka.consumer.auto-offset-reset=earliest

确保反序列化器能正确解析消息，避免因解析失败导致返回null。

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