使用spring-kafka测试中的@EmbeddedKafka测试监听器

Question

我正在尝试在使用@KafkaListener创建的springboot中测试监听器但是监听器总是在localhost:9092上侦听，而不是使用这个嵌入式Kafka

我的听众是这样的：

@Component
@Slf4j
class SomeListener {
private final List<String> receivedMessages = new ArrayList<>();
@KafkaListener(topics = "some-ultra-cool-topic")
public void onKafkaMessage(String theMessage) {
log.info("Message received {}", theMessage);
receivedMessages.add(theMessage);
}
Collection<String> getAll() {
return unmodifiableCollection(receivedMessages);
}
}

像这样的spock测试：

@SpringBootTest
@EmbeddedKafka
@TestPropertySource(properties = ['spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}', 'spring.kafka.consumer.auto-offset-reset=earliest'])
class SomeListenerTest extends Specification {
@Autowired
EmbeddedKafkaBroker embeddedKafkaBroker
@Autowired
SomeListener someListener
void 'should receive message'() {
given:
def sender = new KafkaTemplate<>(new DefaultKafkaProducerFactory<String, String>(KafkaTestUtils.producerProps(embeddedKafkaBroker)))
when:
sender.send('some-ultra-cool-topic', 'first message content')
then:
someListener.all.size() == 1
}
}

我的application.yaml没有配置引导服务器，所以它在春季启动时是默认的。

我可以在日志中看到生产者正在向代理发送消息(每次都在不同的随机端口上启动(。但侦听器总是尝试连接到localhost:9092上的代理

我如何配置它来使用这个嵌入式的？

Answer 1

感谢@sawim的提示

实际问题在测试中。我最终用liborg.awaitility:awaitility做了这个测试

then:
waitAtMost(5, SECONDS)
.untilAsserted({ ->
assertThat(personFacade.findAll(), hasSize(1))
})

第一个例子中的配置有效，但在启动过程中，我可以看到kafka日志试图连接到localhost:9200-似乎我们可以忽略它