CodeGym /Courses /Module 5. Spring /Lecture 270: Testing Asynchronous Processes (Kafka, Rabbi...

Lecture 270: Testing Asynchronous Processes (Kafka, RabbitMQ)

Module 5. Spring

Level 21 , Lesson 9

Available

In the previous lecture we got to know Testcontainers: a tool that lets you spin up real databases and other services (including Kafka and RabbitMQ) in a test environment. This will be a key part of today's practice.

Why testing asynchronous processes is challenging?

Asynchronous processes add their own twist to testing. Unlike "regular synchronous requests", where the result is returned immediately (or not returned if something broke), here we're dealing with:

Message queues: Passing information through data brokers (Kafka, RabbitMQ).
Delays: Sometimes messages are not processed instantly.
Checking multiple components: Producers, consumers, and the broker itself (which can have its own quirks).

Main goals of testing asynchronous processes:

Make sure the message is successfully sent by the producer.
Verify that the consumer processed the message correctly.
Ensure messages are not lost and meet the required delivery guarantees.

Tools for testing Kafka and RabbitMQ

For testing asynchronous processes we'll use:

JUnit 5: For writing tests.
Testcontainers: For running isolated instances of Kafka and RabbitMQ.
Spring Kafka Test: Library for testing Kafka in Spring Boot.
Spring AMQP Test: Library for testing RabbitMQ.
Awaitility: For conveniently waiting for asynchronous operations to complete.

How to test Kafka?

1. Setting up the environment with Testcontainers

Testcontainers is great for running Kafka in a test environment. Let's start with the configuration.


import org.junit.jupiter.api.Test;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.utility.DockerImageName;

import static org.junit.jupiter.api.Assertions.assertNotNull;

public class KafkaTest {

    @Test
    void testKafkaContainer() {
        // Start Kafka with Testcontainers
        KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"));
        kafkaContainer.start();

        // Check that Kafka is running
        assertNotNull(kafkaContainer.getBootstrapServers(), "Kafka is not running");

        kafkaContainer.stop();
    }
}

What did we do?

Started Kafka inside a Docker container.
Checked that Kafka started and we can get its bootstrap server address.

2. Testing the producer

Now let's create a test to check whether messages are sent to Kafka correctly.


import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.beans.factory.annotation.Autowired;
import static org.assertj.core.api.Assertions.assertThat;

@EmbeddedKafka(partitions = 1, topics = {"test-topic"})
public class KafkaProducerTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Test
    void testSendMessage() throws Exception {
        // Send a message to Kafka
        ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key", "message");
        RecordMetadata metadata = kafkaTemplate.send(record).get().getRecordMetadata();

        // Verify that the message was successfully sent
        assertThat(metadata.topic()).isEqualTo("test-topic");
    }
}

What did we do?

Used the embedded Kafka broker (@EmbeddedKafka) for testing so we don't have to bring up test infra manually.
Sent a message using KafkaTemplate and verified it was written to the topic.

3. Testing the consumer

Consumers are our "hard workers" that process messages. We'll check that they work correctly.


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;

@EmbeddedKafka(partitions = 1, topics = {"test-topic"})
public class KafkaConsumerTest {

    @Test
    void testConsumeMessage() {
        // Configure the test consumer
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "true", kafkaBroker);
        DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
        Consumer<String, String> consumer = consumerFactory.createConsumer();

        // Subscribe to the topic and verify the received message
        consumer.subscribe(Collections.singletonList("test-topic"));
        ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, "test-topic");

        assertThat(record.value()).isEqualTo("message");
    }
}

How to test RabbitMQ?

1. Preparing the environment

We'll spin up RabbitMQ using Testcontainers.


import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;

import static org.junit.jupiter.api.Assertions.assertTrue;

public class RabbitMQTest {

    @Test
    void testRabbitMQContainer() {
        RabbitMQContainer rabbitMQContainer = new RabbitMQContainer("rabbitmq:3-management");
        rabbitMQContainer.start();

        assertTrue(rabbitMQContainer.isRunning());
        rabbitMQContainer.stop();
    }
}

2. Testing the producer

Let's verify that a message is sent to a RabbitMQ queue.


import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import static org.junit.jupiter.api.Assertions.assertEquals;

public class RabbitProducerTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void testSendMessage() {
        // Send a message
        String queue = "test-queue";
        String message = "Hello RabbitMQ";
        rabbitTemplate.convertAndSend(queue, message);

        // Check that the message was sent
        String receivedMessage = (String) rabbitTemplate.receiveAndConvert(queue);
        assertEquals(message, receivedMessage);
    }
}

3. Testing the consumer

Let's write a test to verify the consumer.


import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;

import static org.junit.jupiter.api.Assertions.assertTrue;

public class RabbitConsumerTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MessageListenerContainer container;

    @Test
    void testConsumeMessage() {
        // Send a message
        String queue = "test-queue";
        String message = "Hello RabbitMQ";
        rabbitTemplate.convertAndSend(queue, message);

        // Verify that the message was processed
        container.start();
        String receivedMessage = (String) rabbitTemplate.receiveAndConvert(queue);
        assertTrue(receivedMessage.equals(message));
        container.stop();
    }
}

Things to keep in mind

Asynchronicity is tricky: account for delays in message processing. Use libraries like Awaitility to wait.
Don't forget to isolate tests: Testcontainers or @EmbeddedKafka/@RabbitMQ will help with that.
Kafka and RabbitMQ are message queues with different characteristics, so testing approaches might differ slightly.