java EmbeddedKafka在后续测试中延迟后向消费者发送消息
我使用的是卡夫卡弹簧测试2.0.1版中的EmbeddedKafka
。发布(最新版本)。
我有一些非常简单的测试,当只运行一个测试时,它们可以正常工作但每当我一个接一个地运行它们(所以只运行整个测试类),第二个测试就会失败——消费者不会收到任何消息
public class KafkaControllerTest {
private static final String FOO_TOPIC = "fooTopic";
@Rule
public KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, FOO_TOPIC);
MessageConsumer msgConsumerFoo = mock(MessageConsumer.class);
@Before
public void before() {
assertTrue(embeddedKafka.getBrokerAddresses().length == 1);
KafkaController controller = new KafkaController(
embeddedKafka.getBrokerAddress(0).toString(),
new Consumer(FOO_TOPIC, msgConsumerFoo));
MockMvcBuilders.standaloneSetup(controller).build();
}
@Test
public void kafkaFirstTest() throws Exception {
sendMessage(FOO_TOPIC, "foo message");
verify(msgConsumerFoo).consume(any());
}
@Test
public void kafkaSecondTest() throws Exception {
sendMessage(FOO_TOPIC, "foo2 message");
verify(msgConsumerFoo).consume(any());
}
void sendMessage(String topic, String notification) throws ExecutionException, InterruptedException {
Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka);
Producer<Integer, String> producer = new DefaultKafkaProducerFactory<Integer, String>(props).createProducer();
producer.send(new ProducerRecord<>(topic, notification)).get();
}
}
以及测试等级的代码:
public class KafkaController {
public KafkaController(String brokerAddress, Consumer... consumers) {
for (Consumer consumer : consumers) {
addTopicListener(brokerAddress, consumer.topic, consumer.messageConsumer);
}
}
private void addTopicListener(String brokerAddress, String topic, MessageConsumer consumer) {
HashMap<String, Object> consumerConfig = new HashMap<>();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerAddress);
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerConfig,
new StringDeserializer(),
new StringDeserializer());
ContainerProperties containerProperties = new ContainerProperties(topic);
containerProperties.setMessageListener((MessageListener<String, String>) data -> {
consumer.consume(data.value());
});
ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProperties);
container.start();
}
public interface MessageConsumer {
void consume(String message);
}
public static class Consumer {
private final String topic;
private final MessageConsumer messageConsumer;
public Consumer(String topic, MessageConsumer messageConsumer) {
this.topic = topic;
this.messageConsumer = messageConsumer;
}
}
}
我认为问题是由{
我有什么遗漏吗?我没有找到任何有帮助的消费品
It is generally recommended to use the rule as a @ClassRule to avoid starting/stopping the broker between tests (and use a different topic for each test).
然而,我试着把KafkaEmbedded
变成@ClassRule
并在测试中使用不同的主题,但仍然一无所获
这个问题与异步性有关,因为在第二次测试中添加延迟有助于:
@Test
public void kafkaSecondTest() throws Exception {
Thread.sleep(1000);
sendMessage(FOO_TOPIC, "foo2 message");
Thread.sleep(1000);
verify(msgConsumerFoo).consume(any());
}
是的,只有在发送消息之前和之后添加Thread.sleep(1000)
时,它才会起作用
那么,我如何检查KafkaEmbedded
或其他组件是否准备好发送/使用消息呢
# 1 楼答案
您可能需要设置
auto.offset.reset=earliest
(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG
)。否则,您可能会在容器完全启动之前发送消息当然,如果您使用相同的主题,第二个消费者应该会收到两条消息
此外,我建议在每次测试结束时停止容器。调试日志也应该有所帮助
编辑
另一种方法是在测试中发送消息之前使用
ContainerTestUtils.waitForAssignment()
;我们在框架测试中经常这样做