Integration testing with Spring Boot and embedded kafka
Tags: java, kafka, springboot, testing
Published: 2024-06-15 16:21:27


In several projects, I have encountered difficulties in implementing integration tests for Spring Boot applications using Kafka, and developers are often put off by the effort required to implement tests involving Kafka. This post describes the implementation of a simple integration test using an embedded Kafka broker and the test utility code provided by the `spring-kafka-test` dependency, based on a simple example application.
The sample application ingests messages from the `not-enriched-user-data` Kafka topic and then enriches them with data from a database. Finally, the enriched messages are published to the `enriched-user-data Kafka` topic.

![Architecture of simple test application](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vv6iamv9ftri4snp9i2n.png)

You can find the code of the application [here](https://github.com/SteffenWDA/devto-example-integration-test-embedded-kafka#devto-example-integration-test-embedded-kafka).  For the application an integration test consisting of the following steps is implemented.

1. publish test data to Kafka topic `not-enriched-user-data`
2. message is consumed by the Kafka listener
3. application enriches message with data
4. application sends enriched message to the topic `enriched-user-data`
5. verify that topic `enriched-user-data` contains message with expected content

## Implementation Integration test

Before the test case can be implemented, some code must be written to enable the actual test case implementation.
```java
@SpringBootTest
@EmbeddedKafka(ports = 9092)
class EmbeddedKafkaIntegrationTest {


    @Autowired
    KafkaTemplate<String, UserData> kafkaTemplate;

    @Autowired
    ConsumerFactory<String, EnrichedUserData> consumerFactory;

    @Autowired
    AdditionalUserInformationRepository additionalUserInformationRepository;


    @Test
    void executeIntegrationTest() {
    .....
    }

}
```
With the annotation `@SpringBootTest` the Spring Boot application context  is made available during the test execution. From the application context the `KafkaTemplate`, `ConsumerFactory` and  `AdditionalUserInformationRepository` are injected using the `@Autowired` annotation. The annotation `@EmbeddedKafka` is used to start an in memory Kafka instance reachable at port `9092`.
The following code shows the actual implementation of the test case.

```java
    @Test
    void executeIntegrationTest() {
        //arrange
        final String customerNumber = "customerNumber";
        final String userName = "userName";
        final String interestingAdditionalInformation = "interesting additional information";

        AdditionalUserInformation additionalUserInformation = new AdditionalUserInformation();
        additionalUserInformation.setAdditionalInformation(interestingAdditionalInformation);
        additionalUserInformation.setCustomerNumber(customerNumber);
        additionalUserInformationRepository.save(additionalUserInformation);

        Consumer<String, EnrichedUserData> testConsumer = consumerFactory.createConsumer("test", "test");
        testConsumer.subscribe(List.of("enriched-user-data"));

        //act
        kafkaTemplate.send("not-enriched-user-data", new UserData(userName, customerNumber));

        //assert
        ConsumerRecord<String, EnrichedUserData> receivedRecord = KafkaTestUtils.getSingleRecord(testConsumer, "enriched-user-data");
        Assertions.assertAll("",
                () -> assertEquals(userName, receivedRecord.value().getUserName()),
                () -> assertEquals(customerNumber, receivedRecord.value().getCustomerNumber()),
                () -> assertEquals(interestingAdditionalInformation, receivedRecord.value().getEnrichedInfo())
        );

    }
```
First, an `additionalUserInformation` object is built and saved in the database via the injected `additionalUserInformationRepository`. Then the injected `consumerFactory` object is used to create the Kafka consumer `testConsumer` which subscribes to the `enriched-user-data` topic. With the autowired Kafka template object, a message is sent to the `not-enriched-user-data` topic.
The send message is automatically processed by the Kafka listener of the application. The `getSingleRecord` method from the class `KafkaTestUtils` makes the passed consumer `testConsumer` poll the topic `enriched-user-data` until it receives one record. The retrieved record is used  to validate the correct processing of the message.
## Conclusion

The combination of functionality provided  by `KafkaTestUtils` and the embedded Kafka instance allows the implementation of integration tests without  a lot of effort caused by involvement of Kafka. A key advantage of using an embedded Kafka instance is that it does not require the pulling of container images. As a result, execution is faster than test implementations using the Testcontainers framework, and the tests do not require changes to the existing CI/CD infrastructure to enable image pulling during test execution.