Blog

Spring Kafka Test Examples

Key Takeaways

  • Test listener validation and business behavior directly, without starting Spring or Kafka.
  • Use embedded Kafka for focused integration tests that exercise real producers, consumers, serialization, and listener wiring.
  • Let the embedded broker choose available ports, isolate topics and consumer groups, and close clients you create.
  • Wait for an observable processing result: a successful producer send doesn't prove that a listener finished.
  • Test retries, dead-letter delivery, and committed offsets separately from the happy path. Embedded Kafka isn't a substitute for deployment-level testing.

Spring Kafka testing works best in layers: call listener logic directly for unit tests, then use a broker to verify message delivery and Spring configuration. Spring Kafka's embedded-broker utilities provide an in-process integration-test option; they run real broker infrastructure, not an in-memory simulation.

Why Test with Spring Kafka?

A listener can pass every business-logic test and still fail because its topic, serializer, consumer group, or container configuration is wrong. Conversely, starting a broker to test every validation branch makes the suite harder to maintain without improving those assertions.

Spring Kafka provides @EmbeddedKafka for broker lifecycle management, KafkaTestUtils for client configuration and record assertions, and ContainerTestUtils.waitForAssignment for waiting until listener containers have partitions. Kafka's MockProducer and MockConsumer, plus Spring's corresponding factories, offer broker-free alternatives when testing client interactions.

If you need a refresher on listener registration, start with the Spring Boot KafkaListener example. The tests below focus on what that listener actually does.

Set Up the Test Dependencies

These examples target Spring Boot 4 and Spring Kafka 4, using JUnit Jupiter. In an existing Boot-managed Maven project, add the following dependencies; let Boot manage their versions rather than independently pinning Kafka client and broker libraries.

<!-- pom.xml dependency fragment; Spring Boot dependency management required. -->
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

The Kafka test starter includes the general Boot test starter and spring-kafka-test. The examples use the org.junit.jupiter API familiar from JUnit 5; keep your project's managed JUnit version.

Unit Test a Kafka Listener Without a Broker

Keep the listener thin and inject its business dependency. This example validates and normalizes an order identifier before handing it off. Save it as src/main/java/example/kafka/OrderListener.java.

package example.kafka;

import org.springframework.kafka.annotation.KafkaListener;

public class OrderListener {
    private final Handler handler;

    public OrderListener(Handler handler) {
        this.handler = handler;
    }

    @KafkaListener(id = "orders", topics = "${orders.topic}",
            groupId = "${orders.group}")
    public void onMessage(String value) {
        if (value.isBlank()) {
            throw new IllegalArgumentException("Order must not be blank");
        }
        handler.handle(value.strip());
    }

    @FunctionalInterface
    public interface Handler {
        void handle(String orderId);
    }
}

Save the unit test as src/test/java/example/kafka/OrderListenerTest.java. It calls the actual listener method, checks the normalized result, and verifies that invalid input isn't passed to the handler.

package example.kafka;

import java.util.ArrayList;
import java.util.List;

import org.junit.jupiter.api.Test;

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

class OrderListenerTest {
    @Test
    void normalizesValidOrdersAndRejectsBlankOrders() {
        List<String> handled = new ArrayList<>();
        OrderListener listener = new OrderListener(handled::add);

        listener.onMessage("  order-42  ");
        assertThrows(IllegalArgumentException.class,
                () -> listener.onMessage("   "));

        assertEquals(List.of("order-42"), handled);
    }
}

This is an isolated unit test: no Spring context, listener container, or Kafka client is started. It doesn't validate the annotation's topic or group configuration; those belong in the integration test. The annotation supports property placeholders for those settings.

Test Producer-to-Consumer Delivery with Embedded Kafka

Before adding application listeners, a small round-trip test can verify that a producer writes records a consumer can read. This complete test injects the broker supplied by @EmbeddedKafka, creates a unique topic and group, configures string serialization explicitly, and closes both clients.

Save it as src/test/java/example/kafka/KafkaClientTest.java. The boolean consumerProps overload below is the Spring Kafka 4 API; the older overload accepting an auto-commit string is deprecated for removal.

package example.kafka;

import java.time.Duration;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
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.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;

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

@SpringJUnitConfig(KafkaClientTest.Config.class)
@EmbeddedKafka(partitions = 1)
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
class KafkaClientTest {
    @Autowired
    private EmbeddedKafkaBroker broker;

    @Test
    void sendsAndReceivesARecord() throws Exception {
        String topic = "client-" + UUID.randomUUID();
        broker.addTopics(topic);

        Map<String, Object> producerProps = KafkaTestUtils.producerProps(broker);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);

        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
                broker, "client-" + UUID.randomUUID(), false);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
             KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            broker.consumeFromAnEmbeddedTopic(consumer, topic);
            producer.send(new ProducerRecord<>(topic, "key", "value"))
                    .get(10, TimeUnit.SECONDS);

            ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(
                    consumer, topic, Duration.ofSeconds(10));
            assertEquals("key", record.key());
            assertEquals("value", record.value());
        }
    }

    @Configuration(proxyBeanMethods = false)
    static class Config {
    }
}

There are no hard-coded broker ports or listener addresses. The test utilities obtain the connection details from the injected broker, and getSingleRecord polls for exactly one record within the supplied timeout. Try-with-resources closes the producer and consumer even when an assertion fails.

This verifies Kafka client delivery, not your application's listener behavior. That's the next layer.

Integration Test a Spring Boot Kafka Listener

Now send through a real KafkaTemplate and observe the result of OrderListener. The test substitutes a thread-safe queue for the business dependency, so the assertion waits for processing rather than inspecting shared getLastMessage() state.

Test thread: KafkaTemplate.send()
                    |
                    v
          Embedded broker / topic
                    |
                    v
Listener thread: OrderListener -> Handler -> queue
                                                |
Test thread:                  queue.poll(timeout) -> assert

Save this as src/test/java/example/kafka/KafkaListenerTest.java. Boot supplies the Kafka infrastructure, @Import registers the listener, and bootstrapServersProperty explicitly connects Boot's clients to the embedded broker.

package example.kafka;

import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.ContainerTestUtils;
import org.springframework.test.annotation.DirtiesContext;

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

@SpringBootTest(
        classes = KafkaListenerTest.TestApp.class,
        webEnvironment = SpringBootTest.WebEnvironment.NONE,
        properties = {
            "orders.topic=listener-orders",
            "orders.group=listener-orders-group",
            "spring.kafka.consumer.auto-offset-reset=earliest",
            "spring.kafka.consumer.enable-auto-commit=false",
            "spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer",
            "spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer",
            "spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer",
            "spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer"
        })
@EmbeddedKafka(partitions = 1, topics = "listener-orders",
        bootstrapServersProperty = "spring.kafka.bootstrap-servers")
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
class KafkaListenerTest {
    @Autowired
    private KafkaTemplate<String, String> template;

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private BlockingQueue<String> handledOrders;

    @Test
    void processesAnOrderDeliveredThroughKafka() throws Exception {
        var container = registry.getListenerContainer("orders");
        assertNotNull(container);
        ContainerTestUtils.waitForAssignment(container, 1);

        String expected = "order-" + UUID.randomUUID();
        template.send("listener-orders", "key-test", "  " + expected + "  ")
                .get(10, TimeUnit.SECONDS);

        String actual = handledOrders.poll(10, TimeUnit.SECONDS);
        assertNotNull(actual, "Listener did not process the order within 10 seconds");
        assertEquals(expected, actual);
    }

    @Configuration(proxyBeanMethods = false)
    @EnableAutoConfiguration
    @Import(OrderListener.class)
    static class TestApp {
        @Bean
        BlockingQueue<String> handledOrders() {
            return new LinkedBlockingQueue<>();
        }

        @Bean
        OrderListener.Handler orderHandler(BlockingQueue<String> handledOrders) {
            return handledOrders::add;
        }
    }
}

The send future reports the producer's send result, not listener completion. The separate queue poll waits for the normalized business result and returns null if its timeout expires. Waiting for partition assignment also makes listener readiness explicit.

This is a Spring-context integration test, not a unit test. It covers producer-to-listener delivery but replaces the downstream business dependency; use your real application configuration and assert persisted state or an output record when testing the complete application flow.

For structured messages, use your application's actual serializer and deserializer instead of this string-only fixture. See the Spring Boot Kafka consumer JSON example for that application-side setup.

Choose Embedded Kafka, Testcontainers, or an External Cluster

EnvironmentChoose it forBoundary
Embedded KafkaFocused Kafka integration tests managed alongside the test processThe single-broker examples don't exercise deployment networking, replication, or production security.
TestcontainersA disposable broker running a selected container imageRequires a container runtime; one container still isn't a production cluster.
Externally managed test clusterConfigured authentication, authorization, multiple brokers, and deployment-level checksNeeds provisioning, cleanup, and stronger isolation between test runs.

Testcontainers manages Kafka containers independently of the test JVM. For Apache Kafka images, the current class is org.testcontainers.kafka.KafkaContainer; the older org.testcontainers.containers.KafkaContainer is deprecated. Choose the environment based on what the test must prove, not a blanket assumption that containers are always more realistic.

For topology and replication concerns, review Kafka architecture before deciding which scenarios require a multi-broker environment.

Understand Embedded-Broker Test Costs

Embedded Kafka starts broker infrastructure and uses temporary storage in the test process. Startup time and resource use depend on your setup; don't assume a minimal performance impact. Measure your suite, keep business-logic branches in unit tests, and consider sharing a broker only after defining topic isolation and lifecycle ownership.

Keep Streaming Tests Separate

Spring Kafka supports producers and consumers and also offers Kafka Streams integration. Kafka Streams and Apache Flink are separate stream-processing technologies; a passing listener test doesn't validate a Streams topology or a Flink job. Test those processing flows separately.

Test Error Handling and Delivery Behavior

The happy path isn't enough for a listener that changes business state. Add focused scenarios using the same error-handler configuration your application actually runs.

  • Retries: Make the handler throw a retryable exception. For a record listener configured with DefaultErrorHandler and FixedBackOff(0L, 2L), assert three delivery attempts: the original plus two retries. Also test an exception your application classifies as non-retryable.
  • Dead-letter delivery: A dead-letter topic holds records that couldn't be processed. Configure DeadLetterPublishingRecoverer with an explicit destination, exhaust the retries, then consume that topic and assert the original key, payload, and failure metadata—not merely that an exception was logged.
  • Serialization and deserialization: Test producer serialization failures separately from malformed records arriving at the consumer. With ErrorHandlingDeserializer, exercise a delegate-deserializer failure and verify recovery without invoking normal business processing. For JSON consumers, publish malformed raw bytes rather than an already-valid Java object.
  • Offsets and acknowledgments: An offset tracks a consumer group's position. A successful business assertion doesn't prove that position was committed. Inspect committed offsets or restart a consumer with the same group to test redelivery under your configured acknowledgment mode; don't assign a fresh group for this particular scenario.

Common Mistakes

  • Asserting immediately after sending: Wait for a queue result, output record, or persisted business effect with a deadline.
  • Reusing topics without a cleanup strategy: Prefer unique topics and groups when tests share a broker. Resetting to earliest can expose records left by earlier tests.
  • Assuming these fixtures support arbitrary parallel execution: Embedded brokers publish addresses through JVM system properties. Run broker-owning test classes serially or isolate them in separate processes; don't mix global and per-class embedded brokers.
  • Growing a shared observation queue indefinitely: The listener example has one method and a class-owned broker/context. If you add methods, introduce per-test routing or correlation-aware observations instead of relying on test order.
  • Leaking infrastructure: Close manually created clients. Let the Spring test context own managed infrastructure; these examples mark their contexts dirty after the class to make teardown explicit.

FAQ

How do I test a Kafka listener with Spring Kafka?

Call the listener method directly to unit test validation and business behavior. Use a Spring-context test with a broker to verify topic subscription, serialization, listener registration, and delivery.

How do I use @EmbeddedKafka in a JUnit Jupiter test?

Annotate the test class with @EmbeddedKafka. With @SpringJUnitConfig or @SpringBootTest, inject EmbeddedKafkaBroker from the test context; without a Spring context, the extension can supply it as a test-method parameter.

How do I wait for asynchronous Kafka listener processing in a test?

Wait for an observable result with a timeout, such as a BlockingQueue poll, an output record, or persisted state. A completed KafkaTemplate send future doesn't prove that the listener finished processing.

When should I use embedded Kafka versus Testcontainers?

Use embedded Kafka for focused integration tests without a separate container runtime. Use Testcontainers when you want a disposable broker image and process boundary; neither single-broker setup proves production-cluster behavior.

How do I test a Kafka consumer's retry or dead-letter behavior?

Force a retryable processing failure, assert the configured attempt count, and consume the dead-letter destination to verify the recovered record. Test non-retryable exceptions and deserialization failures separately using your application's actual recovery configuration.

Further Reading

Written by Sam Erickson

Software engineer. I write every article on StackChief myself. Author of Mastering the Tech Interview. More about me

Related reading