Unit Testing Kafka Consumers

by | Nov 16, 2016 | Blog, Business, Data Engineering, NP | 16 comments

Unit testing your Kafka code is incredibly important. It’s transporting your most important data. This is especially true for your Consumers. They are the end point for using the data. There are often many different Consumers using the data. You’ll want to unit test all of them.

In a previous post, I showed you how to unit test Producers.

Refactoring Your Consumers

First of all, you’ll need to be able to change your Consumer at runtime. Instead of using the KafkaConsumer object directly, you’ll use the Consumer interface.

public Consumer<String, String> consumer;

You can use whichever method for dependency injection, but I’m making the Consumer public so I can change it from the unit test.

Next, you’ll want to refactor the code for creating your KafkaConsumer. The creation of the KafkaConsumer should be in separate method that won’t get called by your production Consumer code.

You’ll also need to refactor the code that consumes the data from the Consumer object. This code will need to be callable from the unit test. Also, the Consumer object often consumes in an infinite loop (while (true)). You need to refactor the actual consumption code so it doesn’t get stuck in an infinite loop.

Unit Testing Your Consumer

Kafka unit tests of the Consumer code use MockConsumer object. The @Before will initialize the MockConsumer before each test.

MockConsumer<String, String> consumer;

@Before
public void setUp() {
    consumer = new MockConsumer<String, String>(OffsetResetStrategy.EARLIEST);
}

Have you been searching for the best data engineering training? You’ve found it. Sign up for my list so you can get my Professional Data Engineering course.

Once we’ve set the objects up, we can start testing.

@Test
public void testConsumer() throws IOException {
    // This is YOUR consumer object
    MyTestConsumer myTestConsumer = new MyTestConsumer();
    // Inject the MockConsumer into your consumer
    // instead of using a KafkaConsumer
    myTestConsumer.consumer = consumer;

    consumer.assign(Arrays.asList(new TopicPartition("my_topic", 0)));

    HashMap<TopicPartition, Long> beginningOffsets = new HashMap<>();
    beginningOffsets.put(new TopicPartition("my_topic", 0), 0L);
    consumer.updateBeginningOffsets(beginningOffsets);

    consumer.addRecord(new ConsumerRecord<String, String>("my_topic",
                       0, 0L, "mykey", "myvalue0"));
    consumer.addRecord(new ConsumerRecord<String, String>("my_topic", 0,
                       1L, "mykey", "myvalue1"));
    consumer.addRecord(new ConsumerRecord<String, String>("my_topic", 0,
                       2L, "mykey", "myvalue2"));
    consumer.addRecord(new ConsumerRecord<String, String>("my_topic", 0,
                       3L, "mykey", "myvalue3"));
    consumer.addRecord(new ConsumerRecord<String, String>("my_topic", 0,
                       4L, "mykey", "myvalue4"));

    // This is where you run YOUR consumer's code
    // This code will consume from the Consumer and do your logic on it
    myTestConsumer.consume();

    // This just tests for exceptions
    // Somehow test what happens with the consume()
}

We start off by instantiating the Consumer we’re wanting to test. We inject our MockConsumer into the Consumer. Then, the MockConsumer’s topic, partitions, and beginning offsets need to be set up. We send some data with the Consumer. All of the data added by the MockConsumer will be consumed by the Consumer. We call the addRecord() method for every ConsumerRecord we want the Consumer to see. Finally, we consume the data.

A quick note that this test only validates that the Consumer doesn’t throw an exception while processing this data. To verify the actual processing or output, you may need to mock another object or gather the output in a last and run your assertions.

16 Comments

  1. Vijendra Kulhade
    Vijendra Kulhade on June 14, 2017 at 12:22 pm

    I get this exception when I code under test try to consume.
    java.lang.IllegalStateException: Subscription to topics, partitions and pattern are mutually exclusive

    Reply
    • Niraj
      Niraj on June 18, 2018 at 5:13 am

      I faced the same issue, the workaround for this is to assign new TopicPartition(topic, 0) to a variable and use that. Following code works for me:


      TopicPartition topicPartition = new TopicPartition(topic, 0);
      consumer.assign(Arrays.asList(topicPartition));
      HashMap<TopicPartition, Long> beginningOffsets = new HashMap<>();
      beginningOffsets.put(topicPartition, 0L);
      consumer.updateBeginningOffsets(beginningOffsets);
      consumer.addRecord(new ConsumerRecord<>(topic, 0, 0L, "some-ref", "some-order"));

      Reply
  2. Mark
    Mark on November 7, 2017 at 1:35 am

    but there is no consume method in Consumer class. What should we use instead of it?

    Reply
    • Jesse
      Jesse on November 7, 2017 at 12:32 pm

      I added a comment clarifying that MyTestConsumer is your code. The consume method is where your code starts to consume and process your messages.

      Reply
  3. Tom
    Tom on October 23, 2018 at 1:52 am

    Thank you Jesse your blogs have been extremely helpful getting me off the ground. The coverage of my tests is quite low due to differing construction requirements for the mock and concrete implementations but its a start.

    Regarding the Mock Producer and Consumer i understand they are now deprecated I don’t suppose you have any blogs on the new testing framework?

    Thanks again

    Reply
    • Jesse
      Jesse on October 23, 2018 at 6:05 am

      I hadn’t heard about any deprecation and my quick checks didn’t show any deprecations in the JavaDocs. Where are you seeing this?

      Reply
  4. here_d7
    here_d7 on November 14, 2018 at 9:25 pm

    For the first time, thank you about your info.
    I’m really interested about Kafka and I’m trying to do a consumer and producer unit tests right now but I’m having some problems and doubts after read a lot of information about Kafka:

    1)What is the difference between MockConsumer class/MockProducer class and EmbeddedKafka?
    2)I can to do unit test with the both ways…right?
    3)Could I do a unit test without my real consumer? I mean, using the MockConsumer and MockProducer classes..

    Thanks

    Reply
    • Jesse
      Jesse on November 20, 2018 at 9:58 am

      Embedded spins up a Kafka cluster. You could unit test this way, but I’d prefer mocking for unit tests. Yes, you can unit test any real consumer as shown here.

      Reply
  5. Kevin
    Kevin on January 7, 2019 at 10:57 am

    Do you even run the code you write here? I’ve followed a couple of your tutorials and they’ve always been broken…

    Reply
    • Jesse
      Jesse on January 7, 2019 at 12:18 pm

      I just recomplied and reran this exact unit test. Not only did it compile, but the unit test passed. It passed the tests when I first did the post too. I suggest you try again, actually learn Kafka, learn how to read comments in code, or maybe you need to learn how to code. Good luck!

      Reply
  6. Jeff
    Jeff on June 26, 2019 at 5:20 pm

    Hi Jesse, great post I’ve attended your course on Kafka and learned quite a bit.

    I have a quick question, I don’t seem to have consumer.updateBeginningOffsets in my version of Kafka 1.1.0, do you know which versions of Kafka this was added or removed and what the workaround could potentially be?

    Reply
  7. Sumeet
    Sumeet on September 22, 2019 at 7:34 pm

    Hi,
    I’m having same issue of not having MockConsumer updateBeginningOffsets and addRecord(..) !
    Do I need to use specific version of kafka for unit test?

    Thanks

    Reply

Trackbacks/Pingbacks

  1. Unit Testing Kafka | Jesse Anderson - […] the next post, I show you how to unit test a […]
  2. Last week in Stream Processing & Analytics 11/21/2016 | Enjoy IT - SOA, Java, Event-Driven Computing and Integration - […] Unit Testing Kafka Consumers by Jesse Anderson […]

Submit a Comment

Your email address will not be published.

 
Share This