beginner~2h

Testing the Producer

A Kafka producer with no tests is a producer you can't safely refactor. This module covers the three layers of testing used throughout the reference app.

Learning objectives

Beginner: Distinguish a unit test from an integration test for a Kafka producer.
Intermediate: Unit-test a producer's REST endpoint and its Kafka-sending logic by mocking KafkaTemplate.
Advanced: Write an Embedded Kafka integration test that proves a real message was actually produced to a real (in-process) broker.

	Unit test	Integration test
Scope	One class in isolation, collaborators mocked	Real wiring across multiple components (controller → producer → Kafka)
Speed	Milliseconds	Slower — spins up a Spring context and an embedded broker
Catches	Logic bugs in one class	Wiring/serialization/config bugs that only show up when parts talk to each other
In this module	MockMvc controller test, mocked-KafkaTemplate producer test	Full request → Kafka round trip via Embedded Kafka

Both matter, for different reasons — a unit test tells you the controller validates input correctly even if Kafka is completely unreachable; an integration test tells you the actual bytes that land in Kafka are what you think they are.

@WebMvcTest(LibraryEventController.class)
class LibraryEventControllerTest {

    @Autowired private MockMvc mockMvc;
    @MockBean private LibraryEventProducer libraryEventProducer;

    @Test
    void postLibraryEvent_returns201() throws Exception {
        String json = """
            {"book": {"bookId": 1, "bookName": "Kafka in Action", "bookAuthor": "D."}}
            """;

        mockMvc.perform(post("/v1/libraryevent")
                .contentType(MediaType.APPLICATION_JSON)
                .content(json))
            .andExpect(status().isCreated());

        verify(libraryEventProducer, times(1)).sendLibraryEvent(any());
    }

    @Test
    void postLibraryEvent_missingBookName_returns400() throws Exception {
        String json = """
            {"book": {"bookId": 1, "bookAuthor": "D."}}
            """;

        mockMvc.perform(post("/v1/libraryevent")
                .contentType(MediaType.APPLICATION_JSON)
                .content(json))
            .andExpect(status().isBadRequest());

        verifyNoInteractions(libraryEventProducer);
    }
}

◆ Under the hood

@WebMvcTest loads only the web layer (controllers, filters, advice) — not the full application context, and critically, not any real Kafka wiring. @MockBean replaces LibraryEventProducer with a Mockito mock, so this test says nothing about whether Kafka publishing actually works; it only proves the controller calls the producer correctly (or correctly refuses to, on invalid input).

💻 Code example

@WebMvcTest(LibraryEventController.class)
class LibraryEventControllerTest {

    @Autowired private MockMvc mockMvc;
    @MockBean private LibraryEventProducer libraryEventProducer;

    @Test
    void postLibraryEvent_returns201() throws Exception {
        String json = """
            {"book": {"bookId": 1, "bookName": "Kafka in Action", "bookAuthor": "D."}}
            """;

        mockMvc.perform(post("/v1/libraryevent")
                .contentType(MediaType.APPLICATION_JSON)
                .content(json))
            .andExpect(status().isCreated());

        verify(libraryEventProducer, times(1)).sendLibraryEvent(any());
    }

    @Test
    void postLibraryEvent_missingBookName_returns400() throws Exception {
        String json = """
            {"book": {"bookId": 1, "bookAuthor": "D."}}
            """;

        mockMvc.perform(post("/v1/libraryevent")
                .contentType(MediaType.APPLICATION_JSON)
                .content(json))
            .andExpect(status().isBadRequest());

        verifyNoInteractions(libraryEventProducer);
    }
}

@ExtendWith(MockitoExtension.class)
class LibraryEventProducerTest {

    @Mock private KafkaTemplate<String, String> kafkaTemplate;
    @InjectMocks private LibraryEventProducer producer;

    @Test
    void sendLibraryEvent_callsKafkaTemplateWithCorrectTopic() throws Exception {
        LibraryEvent event = new LibraryEvent(null, LibraryEventType.NEW,
            new Book(1, "Kafka in Action", "D."));

        CompletableFuture<SendResult<String,String>> future = new CompletableFuture<>();
        when(kafkaTemplate.send(eq("library-events"), anyString())).thenReturn(future);

        producer.sendLibraryEvent(event);

        verify(kafkaTemplate, times(1)).send(eq("library-events"), anyString());
    }
}

This isolates LibraryEventProducer 's own logic (JSON serialization, choosing sync vs. async, callback handling) from whether Kafka itself is reachable — pure unit-level correctness.

💻 Code example

@ExtendWith(MockitoExtension.class)
class LibraryEventProducerTest {

    @Mock private KafkaTemplate<String, String> kafkaTemplate;
    @InjectMocks private LibraryEventProducer producer;

    @Test
    void sendLibraryEvent_callsKafkaTemplateWithCorrectTopic() throws Exception {
        LibraryEvent event = new LibraryEvent(null, LibraryEventType.NEW,
            new Book(1, "Kafka in Action", "D."));

        CompletableFuture<SendResult<String,String>> future = new CompletableFuture<>();
        when(kafkaTemplate.send(eq("library-events"), anyString())).thenReturn(future);

        producer.sendLibraryEvent(event);

        verify(kafkaTemplate, times(1)).send(eq("library-events"), anyString());
    }
}

Embedded Kafka starts a real, in-memory Kafka broker inside the JVM running your test — same wire protocol, same client behavior as a real cluster, but disposable and fast enough to run on every CI build.

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

    @Autowired private TestRestTemplate restTemplate;
    @Autowired private EmbeddedKafkaBroker embeddedKafkaBroker;

    private Consumer<String, String> consumer;

    @BeforeEach
    void setUp() {
        Map<String, Object> configs = new HashMap<>(KafkaTestUtils.consumerProps("grp1", "true", embeddedKafkaBroker));
        consumer = new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new StringDeserializer())
            .createConsumer();
        embeddedKafkaBroker.consumeFromAllEmbeddedTopics(consumer);
    }

    @AfterEach void tearDown() { consumer.close(); }

    @Test
    void postLibraryEvent_actuallyPublishesToKafka() {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        String json = """
            {"book": {"bookId": 1, "bookName": "Kafka in Action", "bookAuthor": "D."}}
            """;

        ResponseEntity<LibraryEvent> response = restTemplate.exchange(
            "/v1/libraryevent", HttpMethod.POST, new HttpEntity<>(json, headers), LibraryEvent.class);

        assertEquals(HttpStatus.CREATED, response.getStatusCode());

        ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, "library-events");
        assertTrue(record.value().contains("Kafka in Action"));
    }
}

▲ Pitfall

Embedded Kafka is single-process and doesn't model network partitions, multi-broker replication behavior, or realistic latency — it proves correctness of your wiring and serialization, not your reliability configuration (retries, ISR behavior). Don't treat a green Embedded Kafka suite as proof your acks / min.insync.replicas settings are correct in a real cluster.

✓ Quick recap

What does @WebMvcTest deliberately not load? The rest of the Spring context, including any real Kafka wiring — it isolates just the web layer. What does Embedded Kafka give you that a mocked KafkaTemplate can't? Proof that serialization and topic/partition wiring actually work against a real (if in-memory) broker protocol. Is a passing Embedded Kafka test sufficient proof your production reliability config is correct? No — it doesn't model multi-broker replication, network partitions, or realistic latency.

💻 Code example

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

    @Autowired private TestRestTemplate restTemplate;
    @Autowired private EmbeddedKafkaBroker embeddedKafkaBroker;

    private Consumer<String, String> consumer;

    @BeforeEach
    void setUp() {
        Map<String, Object> configs = new HashMap<>(KafkaTestUtils.consumerProps("grp1", "true", embeddedKafkaBroker));
        consumer = new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new StringDeserializer())
            .createConsumer();
        embeddedKafkaBroker.consumeFromAllEmbeddedTopics(consumer);
    }

    @AfterEach void tearDown() { consumer.close(); }

    @Test
    void postLibraryEvent_actuallyPublishesToKafka() {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        String json = """
            {"book": {"bookId": 1, "bookName": "Kafka in Action", "bookAuthor": "D."}}
            """;

        ResponseEntity<LibraryEvent> response = restTemplate.exchange(
            "/v1/libraryevent", HttpMethod.POST, new HttpEntity<>(json, headers), LibraryEvent.class);

        assertEquals(HttpStatus.CREATED, response.getStatusCode());

        ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, "library-events");
        assertTrue(record.value().contains("Kafka in Action"));
    }
}

Want a visual for this concept?

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

Practice quiz

Next Step

Continue to API Documentation with OpenAPI / Swagger →← Back to all Kafka & Microservices chapters