Spring Boot: Testing Messaging

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production; see the section bibliography for the reference material consulted while preparing these pages.

Messaging code is asynchronous and depends on infrastructure a plain @SpringBootTest does not start on its own, so it earns its own testing playbook on top of the generic one. This page assumes Spring Boot 4.1.x, Spring for Apache Kafka 4.1.x, and Spring AMQP 4.1.x, and covers only what is specific to testing Kafka and RabbitMQ code: which Testcontainers modules Spring Boot auto-wires through @ServiceConnection, when an in-process broker is preferable to a real one, the broker-specific test harnesses that replace hand-rolled polling, and how a contract published as AsyncAPI is actually verified in a test. For the general Testcontainers mechanics (@Testcontainers, @Container, choosing between a slice test and a full @SpringBootTest) see Unit and Integration Testing — it is not repeated here.

Testcontainers with @ServiceConnection

@ServiceConnection (org.springframework.boot.testcontainers.service.connection) reads a running container’s connection details and registers the matching *ConnectionDetails bean automatically, the same mechanism Unit and Integration Testing shows for PostgreSQL. For messaging, which connection details interface gets populated depends on the container class, not on any extra configuration:

Testcontainers class Connection details bean

KafkaContainer, ConfluentKafkaContainer, RedpandaContainer

KafkaConnectionDetails

RabbitMQContainer

RabbitConnectionDetails (default), or RabbitStreamConnectionDetails when opted in

Kafka: KafkaContainer, ConfluentKafkaContainer, and RedpandaContainer

Any of the three Kafka-compatible container classes produces a KafkaConnectionDetails bean, so spring.kafka.bootstrap-servers and any other Kafka auto-configuration property needed by the test’s KafkaTemplate/@KafkaListener beans are wired without a manual @DynamicPropertySource:

@Testcontainers
@SpringBootTest
class OrderEventIntegrationTest {

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("apache/kafka:4.1.0"));

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private OrderEventCounter orderEventCounter; // a test-visible collaborator the listener updates

    @Test
    void publishOrderPlaced_isConsumedByListener() {
        kafkaTemplate.send("orders.order-placed", "ORD-1", "{\"orderId\":\"ORD-1\"}");

        await().atMost(Duration.ofSeconds(5))
            .untilAsserted(() -> assertThat(orderEventCounter.received("ORD-1")).isTrue());
    }
}

Swapping KafkaContainer for ConfluentKafkaContainer (a Confluent Platform image, useful when Confluent Schema Registry or other Confluent-only components matter to the test) or RedpandaContainer (a Kafka-API-compatible broker with a much faster startup time) needs no other change — both still produce KafkaConnectionDetails, and the rest of the test above is identical either way. See Spring Boot Reference Documentation — Testcontainers and Testcontainers for Java — Kafka Module for each container’s constructor options and supported image tags.

RabbitMQ: RabbitMQContainer

RabbitMQContainer produces a RabbitConnectionDetails bean by default, wiring spring.rabbitmq.host, port, username, and password for a RabbitTemplate/@RabbitListener test the same way:

@Testcontainers
@SpringBootTest
class OrderNotificationIntegrationTest {

    @Container
    @ServiceConnection
    static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:4.1-management-alpine");

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void publishOrderPlaced_isRoutedToNotificationQueue() {
        rabbitTemplate.convertAndSend("orders.exchange", "orders.placed", "{\"orderId\":\"ORD-1\"}");

        await().atMost(Duration.ofSeconds(5))
            .untilAsserted(() -> assertThat(rabbitTemplate.receive("notifications.queue")).isNotNull());
    }
}

RabbitMQ Streams need an opt-in: RabbitConnectionDetails is what @ServiceConnection produces from a RabbitMQContainer unless the annotation’s type attribute asks for the stream variant explicitly, and the container must publish the stream port for that to work (5552 plain, 5551 when SSL is configured):

@Container
@ServiceConnection(type = RabbitStreamConnectionDetails.class)
static RabbitMQContainer rabbitStream = new RabbitMQContainer("rabbitmq:4.1-management-alpine")
    .withExposedPorts(5552);

See Spring Boot Reference Documentation — Testcontainers and Testcontainers for Java — RabbitMQ Module.

Docker Compose support

spring-boot-docker-compose is Spring Boot’s alternative to hand-declaring @Container fields: it reads a compose.yaml at the project root, starts the services it declares before the context refreshes, and populates the same connection details beans (KafkaConnectionDetails, RabbitConnectionDetails, and so on) from the running containers — so application code and tests see an ordinary Kafka/RabbitMQ connection without either one knowing Docker Compose is involved:

services:
  kafka:
    image: 'apache/kafka:4.1.0'
    ports:
      - '9092:9092'
  rabbit:
    image: 'rabbitmq:4.1-management-alpine'
    ports:
      - '5672:5672'
      - '15672:15672'

With spring-boot-docker-compose on the test classpath and this file present, a plain @SpringBootTest (no @Testcontainers, no @Container fields at all) starts both services automatically and tears them down when the JVM exits, governed by spring.docker.compose.lifecycle-management (start-and-stop by default, start-only when several applications share one compose.yaml and stopping should be a manual docker compose stop). This trades the fine-grained per-class container control @Container fields give for a setup that matches local docker compose up development exactly. See Spring Boot How-to Guides — Docker Compose, and Compose for Local Development for every spring.docker.compose.* property, its default, and how service connections and their labels work.

@EmbeddedKafka vs. a real Kafka container

@EmbeddedKafka starts an in-process broker — EmbeddedKafkaKraftBroker, the only EmbeddedKafkaBroker implementation since Spring for Apache Kafka moved fully to KRaft — inside the test JVM itself, with no Docker daemon involved:

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = "orders.order-placed",
        bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class OrderEventEmbeddedKafkaTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private OrderEventCounter orderEventCounter;

    @Test
    void publishOrderPlaced_isConsumedByListener() {
        kafkaTemplate.send("orders.order-placed", "ORD-1", "{\"orderId\":\"ORD-1\"}");

        await().atMost(Duration.ofSeconds(5))
            .untilAsserted(() -> assertThat(orderEventCounter.received("ORD-1")).isTrue());
    }
}

@EmbeddedKafka starts faster than a container and needs no Docker, which suits fast feedback loops and CI runners without container support. Prefer a real KafkaContainer (or ConfluentKafkaContainer/ RedpandaContainer) instead whenever the test cares about behavior the embedded broker cannot reproduce faithfully: a specific broker version’s wire protocol, SASL/SSL security configuration, multi-broker/ISR behavior, or Confluent-only components such as Schema Registry — Testcontainers runs the actual broker image, the embedded broker runs an in-JVM approximation of one. See Spring for Apache Kafka Reference Documentation — Testing Applications for @EmbeddedKafka’s full configuration surface, including `brokerProperties and KafkaTestUtils.

Messaging-specific test harnesses and contract checks

The building blocks above start real (or embedded) infrastructure; the tools in this section replace hand-rolled polling loops and duplicate listener wiring once that infrastructure is running. For choosing whether a given test needs a Spring context at all — versus a plain Mockito unit test or a test slice — see Unit and Integration Testing.

spring-rabbit-test and RabbitListenerTestHarness

spring-rabbit-test replaces RabbitListenerAnnotationBeanPostProcessor with RabbitListenerTestHarness when the test class carries @RabbitListenerTest, wrapping every @RabbitListener bean in a Mockito spy so a test can verify it was invoked — or block until it is — without writing its own synchronization:

@SpringJUnitConfig
@RabbitListenerTest
class OrderNotificationListenerTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitListenerTestHarness harness;

    @Test
    void onOrderPlaced_invokesListenerWithPayload() {
        rabbitTemplate.convertAndSend("orders.exchange", "orders.placed", "{\"orderId\":\"ORD-1\"}");

        OrderNotificationListener spy = harness.getSpy("orderNotificationListener");
        verify(spy, timeout(5000)).onOrderPlaced("{\"orderId\":\"ORD-1\"}");
    }
}

This only works against a listener method that declares an id on its @RabbitListener annotation (the id the harness looks the spy up by) and that is not final — Mockito cannot spy or advise a final method. @SpringRabbitTest complements this by adding a ready-made CachingConnectionFactory, RabbitTemplate, RabbitAdmin, and listener container factory to the context when a test wants Spring AMQP infrastructure beans without a full application context, and TestRabbitTemplate offers a no-broker alternative that routes send/sendAndReceive calls straight to the matching listener container in-process. See Spring AMQP Reference Documentation — Testing Support.

The Spring Cloud Stream test binder

Applications built on the functional Function/Consumer/Supplier programming model of Spring Cloud Stream (rather than @KafkaListener/@RabbitListener directly) are tested against spring-cloud-stream-test-binder, which replaces the real binder with an in-JVM one and needs no broker, Kafka or RabbitMQ, running at all:

@SpringBootTest
class OrderProcessorTestBinderTest {

    @Autowired
    private InputDestination input;

    @Autowired
    private OutputDestination output;

    @Test
    void orderPlaced_isEnrichedWithStatus() {
        input.send(new GenericMessage<>("{\"orderId\":\"ORD-1\"}".getBytes()), "orderPlaced-in-0");

        Message<byte[]> enriched = output.receive(5000, "orderPlaced-out-0");
        assertThat(enriched.getPayload()).asString().contains("\"status\":\"PLACED\"");
    }

    @SpringBootApplication
    @EnableTestBinder
    static class TestConfiguration {
    }
}

InputDestination.send(Message, String) pushes a message into the named binding as if it had arrived from the real broker; OutputDestination.receive(long, String) polls the binding a Supplier/Function published to, with a timeout instead of a busy loop. This is the messaging-functions equivalent of @WebMvcTest’s `MockMvc: it exercises the actual binding configuration and message conversion without paying for a real broker. See Spring Cloud Stream Reference Documentation — Testing With Test Binder.

Awaitility for asynchronous assertions

Every example above that publishes a message and then checks a side effect uses await().atMost(…​) .untilAsserted(…​) rather than Thread.sleep followed by a direct assertion, because a listener processes a published message on its own container thread at its own pace: the assertion in OrderEventIntegrationTest.publishOrderPlaced_isConsumedByListener above, for instance, would be flaky without it, since nothing guarantees the listener has run by the time kafkaTemplate.send() returns. Awaitility polls the supplied assertion repeatedly until it stops throwing or the timeout elapses, which is both faster than a fixed sleep (it succeeds as soon as the condition is true) and more reliable (it does not fail just because the listener was a little slower than a guessed sleep duration).

Unit-testing Kafka Streams topologies with TopologyTestDriver

A Kafka Streams topology built with StreamsBuilder — whether wired into Spring through StreamsBuilderFactoryBean or built directly — does not need a broker, embedded or otherwise, to be unit-tested: TopologyTestDriver runs the same Topology in-process against in-memory TestInputTopic/ TestOutputTopic stand-ins. This is plain Apache Kafka test infrastructure, not something Spring adds to the topology, so it is covered in full, with a worked word-count example, on Testing with TopologyTestDriver rather than repeated here.

AsyncAPI contract testing

API-First: Messaging documents a topic’s contract — its channel name, its allowed operations, and the Avro schema each message carries — as an AsyncAPI document kept alongside the .avsc schema it references. A contract test closes the loop by asserting the running application actually honors that document instead of trusting it stays in sync by convention: publish onto the exact channel address the AsyncAPI document declares, using the same generated Avro class the referenced schema compiles to, and let deserialization on the consuming side be the check that the two never drifted apart:

@Testcontainers
@SpringBootTest
class OrderPlacedContractTest {

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("apache/kafka:4.1.0"));

    @Autowired
    private KafkaTemplate<String, OrderPlaced> kafkaTemplate; // OrderPlaced: generated from order-placed.avsc

    @Test
    void publishesOnTheChannelAddressDeclaredInTheAsyncApiDocument() {
        // "orders.order-placed.v1" is the channels.orderPlaced.address value in the AsyncAPI document
        OrderPlaced event = OrderPlaced.newBuilder()
            .setOrderId("ORD-1").setCustomerId("cust-1042")
            .setTotalAmount(129.90).setPlacedAt(Instant.now().toEpochMilli())
            .build();

        kafkaTemplate.send("orders.order-placed.v1", event.getOrderId(), event);

        await().atMost(Duration.ofSeconds(5))
            .untilAsserted(() -> assertThat(orderPlacedConsumer.lastReceived()).isEqualTo(event));
    }
}

Because both the producer’s KafkaTemplate<String, OrderPlaced> and the AsyncAPI document’s payload.$ref point at the same order-placed.avsc file, a schema change that is not also reflected in the AsyncAPI document shows up as a Java compilation failure (the generated OrderPlaced class changed shape) long before it would show up as a runtime contract violation — the contract test above is mainly guarding the channel address and routing, which nothing else checks at compile time. See Spring for Apache Kafka Reference Documentation — Testing Applications for the KafkaTemplate/container wiring this test reuses.

References