嵌入式 kafka 生产者测试

Question

我正在编写集成测试来测试 kafka 生产者。

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {kafkaProducerConfig.class, KafkaProducerIT.InnerConfig.class})
@EnableConfigurationProperties(KafkaProducerInfo.class)
@ComponentScan(basePackages = "...")
public class KafkaProducerIT {

    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "testtopic");

    @Autowired
    CustomKafkaProducer<String, String> KafkaProducer;

    @Autowired
    KafkaController kafkaController;

    @Test
    public void whenSendMessage_thenConsumeIt() throws InterruptedException {
        KafkaProducer.produceMessageToKafkaTopic("ahahahwow", "testtopic");
        kafkaController.countDownLatch.await();
    }

    @Configuration
    public static class InnerConfig {

        @Bean
        public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(ConsumerFactory<String, Object> replyConsumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(replyConsumerFactory);
            factory.setBatchListener(true);
            return factory;
        }

        @Bean
        KafkaController kafkaController() {
            return new KafkaController();
        }

    }

    public static class KafkaController {

        CountDownLatch countDownLatch = new CountDownLatch(1);

        @KafkaListener(topics = "testtopic")
        public void listen(final String payload) {
            countDownLatch.countDown();
        }
    }

}

我想向主题发送消息，使用KafkaController 和CountDownLatch 阅读。 我遇到的问题是CountDownLatch 永远不会被触发并且测试只是挂在await 上。

CustomKafkaProducer 只是一个在底层使用常规kafkaTemplate 的包装器。

附言

在调试过程中，有几种情况流进入侦听器并测试通过。所以问题与错误的主题名称等无关。

Answer 1

您需要为消费者设置 auto.offset.reset=earliest。默认值是最新的，因此如果消费者在记录发送后开始，则存在竞争条件。