我编写了一个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代理?任何输入都会有帮助。谢谢。
2条答案
按热度按时间kt06eoxx1#
加里罗素的答案是最好的解决方案。解决此问题的另一种方法是将消息发送步骤延迟一段时间。这将使消费者做好准备。以下也是正确的解决方案。
经验教训-对于单元测试kafka消费者,要么使用测试用例中的所有消息,要么确保消费者在生产者发送消息之前准备就绪。
ruyhziif2#
消息在使用者启动之前发送。
默认情况下,新消费者在主题末尾开始消费。
添加