Я пытаюсь написать модульный тест для слушателя Kafka, который я разрабатываю с помощью Spring Boot 2.x. Будучи модульным тестом, я не хочу запускать полный сервер Kafka как экземпляр Zookeeper. Итак, я решил использовать Spring Embedded Kafka.
Определение моего слушателя очень простое.
@Component
public class Listener {
private final CountDownLatch latch;
@Autowired
public Listener(CountDownLatch latch) {
this.latch = latch;
}
@KafkaListener(topics = "sample-topic")
public void listen(String message) {
latch.countDown();
}
}
Также очень проста проверка, которая проверяет, что счетчик latch равен нулю после получения сообщения.
@RunWith(SpringRunner.class)
@SpringBootTest
@DirtiesContext
@EmbeddedKafka(topics = { "sample-topic" })
@TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}" })
public class ListenerTest {
@Autowired
private KafkaEmbedded embeddedKafka;
@Autowired
private CountDownLatch latch;
private KafkaTemplate<Integer, String> producer;
@Before
public void setUp() {
this.producer = buildKafkaTemplate();
this.producer.setDefaultTopic("sample-topic");
}
private KafkaTemplate<Integer, String> buildKafkaTemplate() {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
return new KafkaTemplate<>(pf);
}
@Test
public void listenerShouldConsumeMessages() throws InterruptedException {
// Given
producer.sendDefault(1, "Hello world");
// Then
assertThat(latch.await(10L, TimeUnit.SECONDS)).isTrue();
}
}
К сожалению, тест не проходит, и я не могу понять почему. Можно ли использовать экземпляр KafkaEmbedded для тестирования метода, отмеченного аннотацией @KafkaListener?
Весь код доступен в моем репозитории GitHub kafka-listener.
Спасибо всем.
