@kafkalistener在单元测试用例中不会从容器工厂消耗

4nkexdtk  于 2021-06-08  发布在  Kafka
关注(0)|答案(2)|浏览(442)

我编写了一个junit测试用例来测试SpringKafka文档中“使用java配置”一课中的代码(https://docs.spring.io/spring-kafka/reference/htmlsingle/#_with_java_configuration). 一个区别是我在类中使用了一个嵌入式kafka服务器,而不是localhost服务器。我使用的是SpringBoot2.0.2及其SpringKafka依赖项。
在运行这个测试用例时,我看到使用者没有从主题中读取消息,“asserttrue”检查失败。没有其他错误。

@RunWith(SpringRunner.class)
public class SpringConfigSendReceiveMessage {

    public static final String DEMO_TOPIC =  "demo_topic";
    @Autowired
    private Listener listener;

    @Test
    public void testSimple() throws Exception {
        template.send(DEMO_TOPIC, 0, "foo");
        template.flush();
        assertTrue(this.listener.latch.await(60, TimeUnit.SECONDS));
    }

    @Autowired
    private KafkaTemplate<Integer, String> template;

    @Configuration
    @EnableKafka
    public static class Config {

        @Bean
        public KafkaEmbedded kafkaEmbedded() {
            return new KafkaEmbedded(1, true, 1, DEMO_TOPIC);
        }

        @Bean
        public ConsumerFactory<Integer, String> createConsumerFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded().getBrokersAsString());
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return new DefaultKafkaConsumerFactory<>(props);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(createConsumerFactory());
            return factory;
        }

        @Bean
        public Listener listener() {
            return new Listener();
        }

        @Bean
        public ProducerFactory<Integer, String> producerFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded().getBrokersAsString());
            props.put(ProducerConfig.RETRIES_CONFIG, 0);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return new DefaultKafkaProducerFactory<>(props);
        }

        @Bean
        public KafkaTemplate<Integer, String> kafkaTemplate() {
            return new KafkaTemplate<Integer, String>(producerFactory());
        }
    }       

}

class Listener {
    public final CountDownLatch latch = new CountDownLatch(1);

    @KafkaListener(id = "foo", topics = DEMO_TOPIC)
    public void listen1(String foo) {
        this.latch.countDown();
    }
}

我认为这是因为@kafkalistener在阅读主题时使用了一些错误的/默认设置。我在日志中没有看到任何错误。
这个单元测试用例正确吗?如何找到为kafkalistener注解创建的对象,并查看它使用哪个kafka代理?任何输入都会有帮助。谢谢。

kt06eoxx

kt06eoxx1#

加里罗素的答案是最好的解决方案。解决此问题的另一种方法是将消息发送步骤延迟一段时间。这将使消费者做好准备。以下也是正确的解决方案。
经验教训-对于单元测试kafka消费者,要么使用测试用例中的所有消息,要么确保消费者在生产者发送消息之前准备就绪。

@Test
public void testSimple() throws Exception {
    Thread.sleep(1000);
    template.send(DEMO_TOPIC, 0, "foo");
    template.flush();
    assertTrue(this.listener.latch.await(60, TimeUnit.SECONDS));
}
ruyhziif

ruyhziif2#

消息在使用者启动之前发送。
默认情况下,新消费者在主题末尾开始消费。
添加

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

相关问题