Spring Kafka Test Examples

Key Takeaways

• Spring Kafka simplifies testing Kafka applications with dedicated testing utilities.
• Use embedded Kafka for realistic testing scenarios without needing a separate Kafka cluster.
• Ensure that your listener configuration is tested thoroughly to catch potential runtime issues.
• Spring Kafka's testing tools support both unit and integration testing effectively.

Why Test with Spring Kafka?

Testing Kafka-based applications can be challenging. Kafka itself is distributed, requiring coordination and management of clusters for testing purposes. That's where Spring Kafka steps in, providing excellent testing utilities that streamline the process, allowing you to ensure your applications behave correctly under various scenarios right within your test suite.

Setting Up Embedded Kafka

Using an embedded Kafka broker for testing is one of the easiest ways to mimic a production-like environment without the overhead of managing real clusters. Here's how you can set up an embedded Kafka broker using Spring Kafka:

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;

import java.util.Map;

@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
class KafkaTemplateTest {

    private static final String TOPIC = "test-topic";

    @BeforeEach
    void setUp() {
        // Configuration and client setup for producer and consumer
        Map producerProps = KafkaTestUtils.producerProps(embeddedKafka);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producer = new KafkaProducer<>(producerProps);

        Map consumerProps = KafkaTestUtils.consumerProps("embedded-test-group", "true", embeddedKafka);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Collections.singleton(TOPIC));
    }

    @Test
    void testKafkaSendReceive() {
        // Send a message
        producer.send(new ProducerRecord<>(TOPIC, "key", "value"));

        // Validate that the message is consumed
        ConsumerRecord singleRecord = KafkaTestUtils.getSingleRecord(consumer, TOPIC);
        assertEquals("value", singleRecord.value());
    }
}

Unit Testing Kafka Listeners

When you want to unit test your Kafka listener, you can utilize Spring's support for testing beans in an isolated setup. Here's an example of testing a listener using a simple Kafka message listener:

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import static org.assertj.core.api.Assertions.assertThat;

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

    @Autowired
    private MyKafkaListener myKafkaListener;

    @Test
    void testListener() {
        // Send a message to the topic
        KafkaTestUtils.send(embeddedKafka, "test-topic", "key-test", "value-test");

        // Assert that listener processed the message
        assertThat(myKafkaListener.getLastMessage()).isEqualTo("value-test");
    }
}

Integration Testing with Spring Kafka

For integration testing, Spring Kafka supports a more comprehensive approach by allowing you to test complete data flows. You set up the test environment with embedded Kafka and verify how different components interact. This includes producer, consumer, and any intermediate processes that might affect message flow.

FAQ

What is the advantage of using an embedded Kafka broker for testing?

An embedded Kafka broker simplifies the setup and teardown processes during testing. It allows you to run tests on a local setup without needing an external Kafka cluster, closely simulating production environments.

Can I use Spring Kafka for real-time streaming data?

Yes, Spring Kafka is capable of handling real-time streaming data. However, for real-time monitoring and more complex streaming scenarios, consider combining it with other tools like Apache Flink or Kafka Streams API.

How do embedded Kafka clusters impact test performance?

Embedded Kafka clusters generally have minimal impact on test performance for small to medium test sizes. However, they may become resource-intensive for larger tests due to their reliance on a JVM in-memory setup.