intermediate~1h

Testing the Consumer with Embedded Kafka

The producer's Embedded Kafka test (Module 07) proved a record lands in Kafka. This module proves the consumer turns that record into correct database state — the assertion that actually matters end to end.

Learning objectives

Beginner: Explain what an Embedded Kafka consumer test proves that a producer-side test alone doesn't.
Intermediate: Set up an Embedded Kafka test that asserts the consumer correctly persists a consumed record to the database.
Advanced: Diagnose a flaky consumer test caused by asserting before the consumer's poll loop has actually processed the message.

@SpringBootTest
@EmbeddedKafka(topics = "library-events", partitions = 3)
@TestPropertySource(properties = {
    "spring.kafka.consumer.bootstrap-servers=${spring.embedded.kafka.brokers}",
    "spring.kafka.producer.bootstrap-servers=${spring.embedded.kafka.brokers}"
})
class LibraryEventsConsumerIntgTest {

    @Autowired private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired private LibraryEventRepository libraryEventRepository;
    @SpyBean private LibraryEventsConsumer libraryEventsConsumerSpy;
    @SpyBean private LibraryEventsService libraryEventsServiceSpy;

    @Test
    void newEvent_getsPersisted() throws Exception {
        String json = """
            {"libraryEventType":"NEW","book":{"bookId":456,"bookName":"Kafka in Action","bookAuthor":"D."}}
            """;
        kafkaTemplate.sendDefault(json).get();

        // awaitility polls until the async consumer has actually processed the record
        Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
            verify(libraryEventsConsumerSpy, times(1)).onMessage(any());
            verify(libraryEventsServiceSpy, times(1)).processLibraryEvent(any());

            List<LibraryEventEntity> events = libraryEventRepository.findAll();
            assertEquals(1, events.size());
            assertEquals("Kafka in Action", events.get(0).getBook().getBookName());
        });
    }
}

◆ Under the hood — why this test needs Awaitility

Unlike a synchronous MockMvc call, publishing to Kafka and having the listener container pick it up is inherently asynchronous — the test thread has no direct signal for "the consumer is done." Awaitility polls the assertion repeatedly until it passes or a timeout elapses, instead of a fixed Thread.sleep() that's either too short (flaky) or too long (slow suite). This pattern recurs anywhere you assert on the effect of a @KafkaListener.

▲ Pitfall

A fixed Thread.sleep(2000) instead of Awaitility is one of the most common sources of flaky Kafka integration tests — it passes reliably on a fast local machine and fails intermittently on a loaded CI runner, because it hardcodes an assumption about processing latency instead of polling for the actual outcome.

✓ Quick recap

Why can't you assert on consumer effects immediately after publishing, the way you can with a synchronous REST call? Because @KafkaListener processing happens on a separate container thread, asynchronously relative to the test thread. What's the problem with Thread.sleep() as a substitute for Awaitility? It hardcodes a latency assumption, making tests either slow (over-long sleep) or flaky (too-short sleep) depending on machine load.

💻 Code example

@SpringBootTest
@EmbeddedKafka(topics = "library-events", partitions = 3)
@TestPropertySource(properties = {
    "spring.kafka.consumer.bootstrap-servers=${spring.embedded.kafka.brokers}",
    "spring.kafka.producer.bootstrap-servers=${spring.embedded.kafka.brokers}"
})
class LibraryEventsConsumerIntgTest {

    @Autowired private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired private LibraryEventRepository libraryEventRepository;
    @SpyBean private LibraryEventsConsumer libraryEventsConsumerSpy;
    @SpyBean private LibraryEventsService libraryEventsServiceSpy;

    @Test
    void newEvent_getsPersisted() throws Exception {
        String json = """
            {"libraryEventType":"NEW","book":{"bookId":456,"bookName":"Kafka in Action","bookAuthor":"D."}}
            """;
        kafkaTemplate.sendDefault(json).get();

        // awaitility polls until the async consumer has actually processed the record
        Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
            verify(libraryEventsConsumerSpy, times(1)).onMessage(any());
            verify(libraryEventsServiceSpy, times(1)).processLibraryEvent(any());

            List<LibraryEventEntity> events = libraryEventRepository.findAll();
            assertEquals(1, events.size());
            assertEquals("Kafka in Action", events.get(0).getBook().getBookName());
        });
    }
}

Want a visual for this concept?

Generate a diagram tailored to “Testing the Consumer with Embedded Kafka” — the AI picks whichever visual (flowchart, comparison, sequence, etc.) best fits.

Practice quiz

Next Step

Continue to CRUD & Exposing Data →← Back to all Kafka & Microservices chapters