Read time: ~

Testing Kafka Applications

EmbeddedKafka vs Testcontainers, testing producers and listeners, asserting on async consumers with Awaitility, testing serialization, and testing streams with TopologyTestDriver.

Kafka code is asynchronous and involves a broker, which makes it easy to write flaky tests. This final production-readiness module shows how to test producers, listeners, and streams reliably: which broker to run in tests, how to assert on asynchronous consumers without sleeping, and how to test a Kafka Streams topology with no broker at all.

---

What you’ll be able to do after this module

• Choose between EmbeddedKafka and Testcontainers Kafka.
• Test a producer and a @KafkaListener end to end.
• Assert on asynchronous consumers with Awaitility, never Thread.sleep.
• Test serialization and schema compatibility.
• Test a Kafka Streams topology with TopologyTestDriver.

---

1. EmbeddedKafka vs Testcontainers

Two ways to give a test a broker.

Approach	What it is	Use when
EmbeddedKafka	An in-JVM broker from spring-kafka-test	Fast unit/slice tests, no Docker
Testcontainers Kafka	A real broker in a Docker container	Higher fidelity, matches production image

EmbeddedKafka is fastest and needs no Docker, good for the bulk of tests. Testcontainers runs the real broker image, which catches issues an embedded broker can miss and matches your production version. Many teams use EmbeddedKafka widely and Testcontainers for a few high-fidelity integration tests.

---

2. Testing with EmbeddedKafka

Annotate the test with @EmbeddedKafka and point Spring’s Kafka properties at it. The embedded broker starts before the test and shuts down after.

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = "orders")
class OrderFlowTest {

    @Autowired
    private KafkaTemplate<String, OrderCreated> kafkaTemplate;

    @Autowired
    private InventoryListener listener; // records what it received

    @Test
    void publishesAndConsumesOrder() {
        kafkaTemplate.send("orders", "order-1",
            new OrderCreated(1L, "widget", new BigDecimal("49.90")));

        // assertion follows in the next section
    }
}

Spring Boot’s @EmbeddedKafka sets the spring.kafka.bootstrap-servers for the context automatically, so the same producer and listener beans from production run against the embedded broker.

---

3. Asserting on async consumers with Awaitility

A listener runs on another thread, so the record arrives after send() returns. Do not Thread.sleep and hope; poll for the expected condition with Awaitility, which retries until it passes or times out.

@Test
void inventoryReceivesOrder() {
    kafkaTemplate.send("orders", "order-1",
        new OrderCreated(1L, "widget", new BigDecimal("49.90")));

    await().atMost(Duration.ofSeconds(5))
        .untilAsserted(() ->
            assertThat(listener.received())
                .extracting(OrderCreated::orderId)
                .contains(1L));
}

This is reliable because it waits only as long as needed and fails fast with a clear assertion. A Thread.sleep(2000) is both slower (always waits the full time) and flakier (may be too short under load).

---

4. Testing serialization and schema compatibility

Serialization bugs surface at runtime unless you test them. Two things are worth asserting:

• Round trip: serialize a DTO and deserialize it back, confirming equality. This catches field and type mismatches in your serializer config.
• Schema compatibility: for Avro from Schema Registry and Data Contracts, test a proposed new schema against the current one with the Avro SchemaCompatibility checker, so a breaking change fails a unit test rather than a production deploy.

@Test
void newSchemaIsBackwardCompatible() {
    Schema current = loadSchema("order-created-v1.avsc");
    Schema proposed = loadSchema("order-created-v2.avsc");

    SchemaCompatibility.SchemaPairCompatibility result =
        SchemaCompatibility.checkReaderWriterCompatibility(proposed, current);

    assertThat(result.getType())
        .isEqualTo(SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE);
}

---

5. Testing Kafka Streams with TopologyTestDriver

A Kafka Streams topology can be tested with no broker at all. TopologyTestDriver runs the topology in memory: you pipe input records to input topics and read output topics synchronously, so tests are fast and deterministic.

@Test
void aggregatesRevenueByCustomer() {
    Topology topology = buildRevenueTopology();
    try (TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
        TestInputTopic<String, PaymentSucceeded> input = driver.createInputTopic(
            "payments", new StringSerializer(), paymentSerializer);
        TestOutputTopic<String, Double> output = driver.createOutputTopic(
            "revenue-by-customer", new StringDeserializer(), doubleDeserializer);

        input.pipeInput("cust-1", new PaymentSucceeded("cust-1", new BigDecimal("10.00")));
        input.pipeInput("cust-1", new PaymentSucceeded("cust-1", new BigDecimal("5.00")));

        assertThat(output.readKeyValuesToMap()).containsEntry("cust-1", 15.0);
    }
}

This is the right way to test the aggregation, windowing, and join logic from Kafka Streams, because it exercises the topology directly without timing or broker flakiness.

---

6. Guided practical

Run this against your test suite.

1. Write an @EmbeddedKafka test that publishes an OrderCreated and asserts the listener received it with Awaitility.
2. Convert any Thread.sleep in a Kafka test to an Awaitility condition and confirm it is faster and stable.
3. Add a serialization round-trip test for OrderCreated.
4. Add an Avro compatibility test that fails when a required field is removed.
5. Write a TopologyTestDriver test for the revenue aggregation and assert the running total.

Next: Section 8, MSK Architecture and High Availability, where the course turns to operations and production troubleshooting.