Testing Spring Kafka Listener in Spring Boot Test with EmbeddedKafka: A Comprehensive Guide

Setting Up EmbeddedKafka

To get started with testing Spring Kafka listeners using EmbeddedKafka, you’ll need to add the following dependencies to your Spring Boot project:

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
  </dependency>
</dependencies>

Next, create a test configuration class that enables EmbeddedKafka:

@Configuration
@EnableKafka
public class KafkaTestConfig {
  
  @Bean
  public EmbeddedKafkaBroker embeddedKafkaBroker() {
    return new EmbeddedKafkaBroker(1);
  }
  
  @Bean
  public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(embeddedKafkaBroker()));
  }
}

Writing Tests for Spring Kafka Listeners

Now that you have EmbeddedKafka set up, let’s write some tests for your Spring Kafka listeners. Assume you have a listener that consumes messages from a topic called “my_topic”:

@Component
public class MyListener {
  
  @KafkaListener(topics = "my_topic")
  public void onMessage(String message) {
    // process message
  }
}

Create a test class that injects the `MyListener` instance and the `KafkaTemplate`:

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {KafkaTestConfig.class, MyListener.class})
public class MyListenerTest {
  
  @Autowired
  private MyListener myListener;
  
  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;
  
  @Before
  public void setUp() {
    kafkaTemplate.createTopic("my_topic");
  }
  
  @Test
  public void testOnMessage() {
    // send a message to the topic
    kafkaTemplate.send("my_topic", "Hello, World!");
    
    // verify that the message was processed correctly
    // ...
  }
}

Testing Message Consumption

In the `testOnMessage()` method, you can verify that the message was consumed correctly by checking the output of your listener. For example, if your listener writes the message to a database, you can check that the message was inserted correctly:

@Test
public void testOnMessage() {
  // send a message to the topic
  kafkaTemplate.send("my_topic", "Hello, World!");
  
  // verify that the message was inserted into the database
  List<String> messages = dbContext.getMessageRepository().findAll();
  assertThat(messages).contains("Hello, World!");
}

Testing Message Processing

To test that your listener is processing messages correctly, you can use a mock object to verify that the expected method was called:

@Test
public void testOnMessage() {
  // create a mock object for the message processor
  MessageProcessor messageProcessor = mock(MessageProcessor.class);
  
  // set the mock object on the listener
  myListener.setMessageProcessor(messageProcessor);
  
  // send a message to the topic
  kafkaTemplate.send("my_topic", "Hello, World!");
  
  // verify that the message processor was called with the correct argument
  verify(messageProcessor).processMessage("Hello, World!");
}

Advanced Testing Techniques

In addition to testing message consumption and processing, you can also test more advanced scenarios, such as:

Testing Error Handling

To test error handling, you can simulate an exception being thrown by your listener and verify that the error is handled correctly:

@Test
public void testOnError() {
  // set up an error scenario
  when(myListener.onError(any())).thenThrow(new RuntimeException("Error!"));
  
  // send a message to the topic
  kafkaTemplate.send("my_topic", "Hello, World!");
  
  // verify that the error was handled correctly
  // ...
}

Testing Concurrency

To test concurrency, you can use multiple threads to send messages to the topic and verify that the messages are processed correctly:

@Test
public void testConcurrency() {
  // create multiple threads to send messages
  ExecutorService executor = Executors.newFixedThreadPool(5);
  
  for (int i = 0; i < 5; i++) {
    executor.submit(() -> kafkaTemplate.send("my_topic", "Message " + i));
  }
  
  // verify that all messages were processed correctly
  // ...
}

Conclusion

In this article, we’ve covered how to test Spring Kafka listeners using EmbeddedKafka in Spring Boot tests. By following these instructions, you can ensure that your listeners are working correctly and reduce the risk of downstream issues. Remember to test message consumption, processing, and advanced scenarios like error handling and concurrency to ensure that your listeners are robust and reliable.

By testing your Spring Kafka listeners, you can ensure that your application is functioning as expected and reduce the risk of errors and downtime. Happy testing!

Frequently Asked Questions

Get answers to your burning questions about testing Spring Kafka Listener in Spring Boot test with EmbeddedKafka!