Kafka MockConsumer and MockProducer Example (JUnit 6)

In this Kafka MockConsumer and MockProducer example, we will unit test a Kafka producer and a Kafka consumer without starting a Kafka broker. MockProducer and MockConsumer are test versions of the real Kafka clients, and they come with the kafka-clients library. With them, a test can check what our code sends and give our code the records it should read.

This tutorial tests a small book order service with Java 17, Kafka clients 4.4.0 and JUnit 6.1.3, and every test runs in under a second with mvn test.

1. Prerequisites

We need three things before we start:

1. JDK 17 or newer, because JUnit 6 needs Java 17 at least. Run java -version to check.
2. Maven 3.9 or newer. Run mvn -v to check.
3. A basic idea of what a Kafka producer and a Kafka consumer do. You don’t need Kafka installed, and you don’t need Docker.

2. Create the Project

Let’s start with creating an empty Maven project and adding the three dependencies the example needs. We create a folder named kafka-mockconsumer-mockproducer and put this pom.xml in it:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.howtodoinjava</groupId>
  <artifactId>kafka-mockconsumer-mockproducer</artifactId>
  <version>1.0.0</version>

  <properties>
    <maven.compiler.release>17</maven.compiler.release>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>4.4.0</version>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>2.20.1</version>
    </dependency>
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter</artifactId>
      <version>6.1.3</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-surefire-plugin</artifactId>
        <version>3.5.4</version>
      </plugin>
    </plugins>
  </build>
</project>

The dependencies, and what each one gives us, are listed below.

Dependency	What it gives us
kafka-clients	The real KafkaProducer and KafkaConsumer, and also MockProducer and MockConsumer in the same jar
jackson-databind	Converts a BookOrderEvent to JSON and back
junit-jupiter	JUnit 6: the @Test annotation, the assertions and the engine that runs the tests

Note that there is no separate test library for the mocks. MockProducer and MockConsumer live in the org.apache.kafka.clients.producer and org.apache.kafka.clients.consumer packages of kafka-clients.

The finished project has three classes and two test classes, in the standard Maven layout:

kafka-mockconsumer-mockproducer
├── pom.xml
└── src
    ├── main/java/com/howtodoinjava/kafkamock
    │   ├── model/BookOrderEvent.java
    │   ├── producer/BookOrderPublisher.java
    │   └── consumer/BookOrderAggregator.java
    └── test/java/com/howtodoinjava/kafkamock
        ├── producer/BookOrderPublisherTest.java
        └── consumer/BookOrderAggregatorTest.java

3. The Code Under Test

The example is a small book order service. A BookOrderPublisher sends one event to the book-order-events topic for every order, and a BookOrderAggregator reads those events and adds up how many copies of each book (each ISBN) were ordered.

The event is a Java record (model/BookOrderEvent.java), which Jackson converts to and from JSON:

package com.howtodoinjava.kafkamock.model;

public record BookOrderEvent(String orderId, String isbn, int quantity) {
}

The publisher (producer/BookOrderPublisher.java) turns the event into JSON and sends it with the order id as the record key:

package com.howtodoinjava.kafkamock.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.howtodoinjava.kafkamock.model.BookOrderEvent;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.concurrent.Future;

public class BookOrderPublisher {

    public static final String TOPIC = "book-order-events";

    private final Producer<String, String> producer;
    private final ObjectMapper objectMapper = new ObjectMapper();

    // The publisher receives a Producer instead of creating a KafkaProducer itself,
    // so a test can pass a MockProducer here
    public BookOrderPublisher(Producer<String, String> producer) {
        this.producer = producer;
    }

    public Future<RecordMetadata> publish(BookOrderEvent event)
            throws JsonProcessingException {
        String json = objectMapper.writeValueAsString(event);
        return producer.send(new ProducerRecord<>(TOPIC, event.orderId(), json));
    }
}

The aggregator (consumer/BookOrderAggregator.java) polls once and keeps a running total per ISBN:

package com.howtodoinjava.kafkamock.consumer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.howtodoinjava.kafkamock.model.BookOrderEvent;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

public class BookOrderAggregator {

    private final Consumer<String, String> consumer;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final Map<String, Integer> quantityByIsbn = new HashMap<>();

    // Same idea as the publisher: a test passes a MockConsumer here
    public BookOrderAggregator(Consumer<String, String> consumer) {
        this.consumer = consumer;
    }

    // Polls once and adds the quantity of every order to the running total for its ISBN
    public Map<String, Integer> consumeOnce() throws JsonProcessingException {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> record : records) {
            BookOrderEvent event =
                    objectMapper.readValue(record.value(), BookOrderEvent.class);
            quantityByIsbn.merge(event.isbn(), event.quantity(), Integer::sum);
        }
        return quantityByIsbn;
    }
}

The one design choice that makes these classes testable is in their constructors. Both classes receive a Kafka client instead of creating one. The publisher takes any Producer (the interface that KafkaProducer implements), and the aggregator takes any Consumer. In production, we pass a real KafkaProducer and KafkaConsumer. In a test, we pass a MockProducer and a MockConsumer, which implement the same interfaces.

If a class creates its own KafkaProducer inside its constructor, a test cannot swap it out, and the test needs a running broker. In that case, move the new KafkaProducer<>(…) call out of the class: the code that creates the publisher builds the real producer with its usual properties and passes it to the constructor.

4. Testing the Producer with MockProducer

MockProducer records every record our code sends instead of sending it over the network. The test then reads that list with history() and checks it. Sometimes, we also want to see what our code does when Kafka fails. In such cases, we tell MockProducer to fail the next send.

Step 1: Create the MockProducer

Let’s first create the mock producer. In Kafka clients 4.x, the constructor takes four arguments:

MockProducer<String, String> producer =
        new MockProducer<>(true, null, new StringSerializer(), new StringSerializer());

The four arguments are listed below.

Argument	Value here	What it does
autoComplete	true	Every send() succeeds right away. With false, a send stays pending until the test calls completeNext() or errorNext()
partitioner	null	Picks the partition for each record. null uses Kafka’s default
keySerializer	StringSerializer	Turns the key into bytes, the same as the real producer would
valueSerializer	StringSerializer	Turns the value into bytes

The trap here comes from older tutorials. They create the mock with three arguments, new MockProducer<>(true, new StringSerializer(), new StringSerializer()). Kafka clients 4.x removed that constructor, so a test class with that line (here named OldCtorCheck) doesn’t compile:

[ERROR] OldCtorCheck.java:[8,13] cannot infer type arguments for org.apache.kafka.clients.producer.MockProducer<>

The error doesn’t say “constructor removed”, which makes it easy to miss. The fix is to add the partitioner argument (null for the default), as in the code above.

Step 2: Send an Event and Check history()

Next, we write the test for a successful send in src/test/java/com/howtodoinjava/kafkamock/producer/BookOrderPublisherTest.java. The test passes the mock to the publisher, publishes one order and then checks what the mock received:

package com.howtodoinjava.kafkamock.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.howtodoinjava.kafkamock.model.BookOrderEvent;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;

class BookOrderPublisherTest {

    @Test
    void publishSendsTheOrderAsJson() throws Exception {
        // autoComplete = true: every send() succeeds immediately
        MockProducer<String, String> producer =
                new MockProducer<>(true, null, new StringSerializer(), new StringSerializer());
        BookOrderPublisher publisher = new BookOrderPublisher(producer);

        BookOrderEvent order = new BookOrderEvent("ORD-1001", "978-0134685991", 2);
        RecordMetadata metadata = publisher.publish(order).get();

        assertEquals("book-order-events", metadata.topic());
        assertEquals(1, producer.history().size());

        ProducerRecord<String, String> sent = producer.history().get(0);
        assertEquals("ORD-1001", sent.key());
        String expectedJson =
                "{\"orderId\":\"ORD-1001\",\"isbn\":\"978-0134685991\",\"quantity\":2}";
        assertEquals(expectedJson, sent.value());
    }
}

publish() returns a Future, and get() waits for its result. Because autoComplete is true, the future is already complete, so get() returns at once with the RecordMetadata (the topic, partition and offset the record was written to).

history() returns every ProducerRecord sent so far, in order. Here it holds one record, and we check its key and its exact JSON value. Checking the whole JSON string catches a renamed field, which a contains() check on one value would miss.

Step 3: Simulate a Failed Send

Now we test what happens when the broker doesn’t answer. We add a second test method to BookOrderPublisherTest, and this time we create the mock with autoComplete set to false, so the send stays pending until the test decides how it ends:

@Test
void publishReportsABrokerFailure() throws JsonProcessingException {
    // autoComplete = false: send() waits for completeNext() or errorNext()
    MockProducer<String, String> producer =
            new MockProducer<>(false, null, new StringSerializer(), new StringSerializer());
    BookOrderPublisher publisher = new BookOrderPublisher(producer);

    BookOrderEvent order = new BookOrderEvent("ORD-1002", "978-0321356680", 1);
    Future<RecordMetadata> result = publisher.publish(order);
    assertFalse(result.isDone());

    producer.errorNext(new TimeoutException("broker did not respond"));

    ExecutionException thrown = assertThrows(ExecutionException.class, result::get);
    assertInstanceOf(TimeoutException.class, thrown.getCause());
}

Right after publish(), the future is not done, because nothing has completed it yet. errorNext() then fails the oldest pending send with the exception we pass, here a TimeoutException (the error a real producer reports when the broker doesn’t respond in time).

The future’s get() then throws an ExecutionException, and its cause is our TimeoutException. The real KafkaProducer reports a failed send the same way, so this test checks our error handling without unplugging a real broker.

5. Testing the Consumer with MockConsumer

MockConsumer works the other way around. Instead of recording what our code sends, it hands our code the records the test gives it. The test adds records with addRecord(), and the next poll() returns them.

Step 1: Assign a Partition Before Adding Records

Let’s start with the test class, src/test/java/com/howtodoinjava/kafkamock/consumer/BookOrderAggregatorTest.java, and the rule that trips up most first attempts. A MockConsumer only returns records for partitions it owns, so the test must assign a partition before it adds any records:

package com.howtodoinjava.kafkamock.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.util.List;
import java.util.Map;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

class BookOrderAggregatorTest {

    private static final String TOPIC = "book-order-events";
    private static final TopicPartition PARTITION = new TopicPartition(TOPIC, 0);

    private MockConsumer<String, String> consumer;
    private BookOrderAggregator aggregator;

    @BeforeEach
    void setUp() {
        consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        aggregator = new BookOrderAggregator(consumer);
    }

    @Test
    void addRecordBeforeAssignThrows() {
        ConsumerRecord<String, String> record =
                new ConsumerRecord<>(TOPIC, 0, 0L, "ORD-2003", "{}");

        IllegalStateException thrown = assertThrows(IllegalStateException.class,
                () -> consumer.addRecord(record));

        assertEquals("Cannot add records for a partition that is not assigned to the consumer",
                thrown.getMessage());
    }
}

The constructor argument OffsetResetStrategy.EARLIEST tells the mock to start reading from the beginning of a partition, the same as auto.offset.reset=earliest on a real consumer.

The test above shows the trap. When addRecord() is called before assign(), the mock throws an IllegalStateException with the message “Cannot add records for a partition that is not assigned to the consumer”. The fix is to call assign() first. Then we call updateBeginningOffsets(), which tells the mock that the partition starts at offset 0. If we skip that call, the first poll() throws an IllegalStateException: “MockConsumer didn’t have beginning offset specified, but tried to seek to beginning”.

Step 2: Add Records and Poll

Next, we add a test method to BookOrderAggregatorTest that feeds two orders for the same book into the mock and checks the total:

@Test
void consumeOnceAddsUpQuantitiesPerIsbn() throws Exception {
    consumer.assign(List.of(PARTITION));
    consumer.updateBeginningOffsets(Map.of(PARTITION, 0L));

    consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 0L, "ORD-2001",
            "{\"orderId\":\"ORD-2001\",\"isbn\":\"978-0134685991\",\"quantity\":3}"));
    consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 1L, "ORD-2002",
            "{\"orderId\":\"ORD-2002\",\"isbn\":\"978-0134685991\",\"quantity\":1}"));

    Map<String, Integer> totals = aggregator.consumeOnce();

    assertEquals(Map.of("978-0134685991", 4), totals);
}

Each ConsumerRecord takes the topic, the partition, the offset, the key and the value. The offsets must go up (0, then 1), as they would in a real partition.

consumeOnce() calls poll() once, and the mock returns both records in that call. The aggregator adds 3 and 1 for ISBN 978-0134685991, so the total is 4.

Step 3: Simulate a Poll Failure

Finally, we add a third test method that checks a broker error on poll() reaches the caller and is not swallowed:

@Test
void consumeOncePassesOnAPollFailure() {
    consumer.assign(List.of(PARTITION));
    consumer.updateBeginningOffsets(Map.of(PARTITION, 0L));
    consumer.setPollException(new KafkaException("broker unavailable"));

    KafkaException thrown = assertThrows(KafkaException.class, aggregator::consumeOnce);

    assertEquals("broker unavailable", thrown.getMessage());
}

setPollException() makes the next poll() throw the exception we pass. Our aggregator doesn’t catch it, so the test expects a KafkaException with the same message. A consumer that catches the error and keeps running would get a different assertion here, such as checking that the totals did not change.

6. Run the Tests

Now we run all five tests from the project folder:

mvn test

The test part of the output looks like this:

[INFO]  T E S T S
[INFO] Running com.howtodoinjava.kafkamock.consumer.BookOrderAggregatorTest
[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0 -- in com.howtodoinjava.kafkamock.consumer.BookOrderAggregatorTest
[INFO] Running com.howtodoinjava.kafkamock.producer.BookOrderPublisherTest
[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0 -- in com.howtodoinjava.kafkamock.producer.BookOrderPublisherTest
[INFO] Results:
[INFO] Tests run: 5, Failures: 0, Errors: 0, Skipped: 0
[INFO] BUILD SUCCESS

All five tests pass, and no broker was started. You will also see three lines starting with “SLF4J: Failed to load class” before the tests. They come from the logging API that kafka-clients uses, and they only mean that no logging library is on the classpath. They don’t affect the tests.

7. FAQs

7.1. Do MockProducer and MockConsumer need a running Kafka broker?

No. Both mocks keep everything in memory and never open a network connection. That is why the tests in this article run in well under a second, on a laptop or in CI, with no Kafka or Docker installed.

7.2. When should I use Testcontainers or an embedded Kafka instead?

Use the mocks to test your own logic: what you send, how you read records and how you react to errors. Use a real broker (Testcontainers starts one in Docker) when you need to test what the mocks skip, such as serializer settings, topic configuration, consumer group rebalancing or exactly-once transactions.

7.3. Does MockProducer check that the topic exists?

No. MockProducer accepts any topic name, so a typo in the topic still passes. Keep the topic name in one constant (like TOPIC in BookOrderPublisher) and assert it in the test, as publishSendsTheOrderAsJson() does with metadata.topic().

7.4. What changes in these tests between JUnit 5 and JUnit 6?

Nothing in the test code. The annotations and assertions used here (@Test, @BeforeEach, assertEquals, assertThrows, assertInstanceOf) are the same in both. Only the dependency version changes, and JUnit 6 needs Java 17 or newer.

8. Conclusion

MockProducer and MockConsumer let us unit test Kafka code without a broker, as long as our classes receive a Producer or Consumer instead of creating one. MockProducer records what we send in history() and fails a send on demand with errorNext(). MockConsumer returns the records we add with addRecord() and fails a poll with setPollException().

The two traps to remember are both about setup. Kafka clients 4.x needs the four-argument MockProducer constructor, and a MockConsumer needs assign() and updateBeginningOffsets() before the first addRecord().

9. References

• MockProducer Javadoc (Kafka 4.x)
• MockConsumer Javadoc (Kafka 4.x)
• JUnit User Guide
• Testcontainers Kafka module

Drop me your questions related to Kafka MockConsumer and MockProducer in the comments.

Happy Learning !!

Source Code on Github