Kafka Integration Testing with @EmbeddedKafka
Fast recall
Quick revision
Core answer
Practical ability to write Kafka integration tests without external dependencies
Using Thread.sleep() to wait for message consumption — use CountDownLatch or Awaitility instead
Mention testing error handling paths and DLT routing, not just happy path
Interview-ready guide
Interview answer
@EmbeddedKafka starts a real Kafka broker inside the test JVM — no Docker, no external infrastructure. Your producer and consumer use the actual Kafka protocol, catching serialization and partition issues that mocks miss.
The main challenge is asynchronicity. The producer sends on one thread, the consumer processes on another. Use `CountDownLatch` or `Awaitility` to wait for the listener to process the message.
For producer tests: send with KafkaTemplate, consume with a test consumer, assert on key/value/partition.
For consumer tests: produce a test message, wait for the @KafkaListener to process it via CountDownLatch.
For error handling tests: send a message that causes failure, verify it reaches the DLT after retries.
Use unique topic names per test to prevent interference — shared topics cause flaky tests.
In practice, I use @EmbeddedKafka for integration tests and Testcontainers with a real Kafka container for more realistic end-to-end tests.
Kafka Integration Testing with @EmbeddedKafka
Direct Answer
@EmbeddedKafka starts an in-memory Kafka broker for tests. Use EmbeddedKafkaBroker to configure test topics, KafkaTemplate to produce test messages, and consumer.poll() or a test listener with CountDownLatch to verify consumption.
Code Example
@EmbeddedKafka spins up an in-process Kafka broker on a random port for the duration of the test class. Real Kafka client code (producers, listeners, error handlers) runs unchanged — no mocks.
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
@SpringBootTest
@EmbeddedKafka(
partitions = 3,
topics = {"orders", "orders.DLT"},
brokerProperties = {
"listeners=PLAINTEXT://localhost:9092",
"port=9092"
})
@DirtiesContext // fresh broker between test classes
class OrderListenerTest {
@Autowired private KafkaTemplate<String, OrderEvent> template;
@Autowired private OrderRepository orderRepo;
@Autowired private EmbeddedKafkaBroker embeddedKafka;
@Test
void listener_persists_order_event() throws Exception {
OrderEvent event = new OrderEvent(UUID.randomUUID(), "alice", new BigDecimal("99.90"));
template.send("orders", event.orderId().toString(), event).get(5, TimeUnit.SECONDS);
// Wait for the listener to consume + persist (avoid Thread.sleep!)
Awaitility.await()
.atMost(Duration.ofSeconds(5))
.untilAsserted(() ->
assertThat(orderRepo.findById(event.orderId())).isPresent());
}
@Test
void poison_message_routes_to_dlt() {
// Send malformed payload → ErrorHandler routes to orders.DLT
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
"dlt-test", "true", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
try (Consumer<String, byte[]> dltConsumer =
new DefaultKafkaConsumerFactory<String, byte[]>(consumerProps,
new StringDeserializer(), new ByteArrayDeserializer()).createConsumer()) {
embeddedKafka.consumeFromAnEmbeddedTopic(dltConsumer, "orders.DLT");
template.send("orders", "bad-id", null); // null payload triggers error
ConsumerRecord<String, byte[]> dlt =
KafkaTestUtils.getSingleRecord(dltConsumer, "orders.DLT", Duration.ofSeconds(5));
assertThat(dlt).isNotNull();
}
}
}
# Override bootstrap servers in test profile so app uses the embedded broker
spring:
config:
activate.on-profile: test
kafka:
bootstrap-servers: ${spring.embedded.kafka.brokers}
@EmbeddedKafkaruns an actual Kafka broker (KRaft or ZK depending on version) — your producer/consumer code is exercised end-to-end without external infrastructure.- Use Awaitility or
CountDownLatchinstead ofThread.sleepto wait for the listener — sleeps are flaky and slow. @DirtiesContextrebuilds the application context (and the broker) between test classes, which is slow. Group related tests in one class to amortize the broker startup cost (~3-5 seconds).- For higher fidelity (real protocol, real disks), use Testcontainers with
confluentinc/cp-kafka— slower (10-15 s startup) but closer to production. Many teams use@EmbeddedKafkafor unit-style tests and Testcontainers for end-to-end suites.
Common Mistakes & Pitfalls
Using Thread.sleep() to wait for message consumption — use CountDownLatch or Awaitility instead
Check your understanding
Answer these without looking back at the article.
- How do you test Kafka transactional producers with @EmbeddedKafka?
- What is the difference between @EmbeddedKafka and Testcontainers for Kafka testing?
- How do you verify message headers in Kafka integration tests?
Related interview prep categories
2Want a guided curriculum order?
These modules are also sequenced in our flagship backend interview track — open it when you prefer an ordered path instead of browsing by topic.