title: Writing JUnit tests for Kafka Consumer
body: <p>I have a kafka consumer which is subscribing on a topic. Implementation is working fine. But when trying to implement unit tests for that, there's a problem because of it's implementing by <code>Runnable</code> interface. </p>
  
  <p><strong>Implementation</strong></p>
  
  <pre><code>@Override
  public void run() {
      kafkaConsumer.subscribe(kafkaTopics);
  
      while (true) {
          ConsumerRecords&lt;String, String&gt; records = kafkaConsumer.poll(1000);
          Map&lt;String, InventoryStock&gt; skuMap = new LinkedHashMap&lt;&gt;();
  
          try {
              // populating sku map with consumer record
              for (ConsumerRecord&lt;String, String&gt; record : records) {
                  populateMap(skuMap, record.value());
              }
  
              if (MapUtils.isNotEmpty(skuMap)) {
                  // writing sku inventory with populated sku map
                  inventoryDao.updateInventoryTable(INVENTORY_JOB_ID, skuMap);
              }
          } catch (Exception e) {
  
          }
          kafkaConsumer.commitAsync();
      }
  }
  </code></pre>
  
  <p>I tried to implement the tests using <code>MockConsumer</code>. But it needs to be assigned to the consumer in implementation. But consumer in implementation doesn't expose out side. Here what I tried. </p>
  
  <pre><code>@Before
  public void onBefore() {
      MockitoAnnotations.initMocks(this);
  
      Properties consumerProps = new Properties();
      consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
      consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
      consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  
      consumer = new MockConsumer&lt;&gt;(OffsetResetStrategy.EARLIEST);
      skuInventoryConsumer = new SkuInventoryConsumer(consumerProps);
  
      KafkaConsumer kafkaConsumerMock = mock(KafkaConsumer.class);
  
      Whitebox.setInternalState(skuInventoryConsumer, "LOGGER", LOGGER);
      Whitebox.setInternalState(skuInventoryConsumer, "kafkaConsumer", kafkaConsumerMock);
  
  }
  
  @Test
  public void should_subscribe_on_topic() {
      consumer.assign(Arrays.asList(new TopicPartition("my_topic", 0)));
  
      HashMap&lt;TopicPartition, Long&gt; beginningOffsets = new HashMap&lt;&gt;();
      beginningOffsets.put(new TopicPartition("my_topic", 0), 0L);
      consumer.updateBeginningOffsets(beginningOffsets);
  
      consumer.addRecord(new ConsumerRecord&lt;&gt;("my_topic", 0, 0L, "mykey", "myvalue0"));
      consumer.addRecord(new ConsumerRecord&lt;&gt;("my_topic", 0, 1L, "mykey", "myvalue1"));
      consumer.addRecord(new ConsumerRecord&lt;&gt;("my_topic", 0, 2L, "mykey", "myvalue2"));
      consumer.addRecord(new ConsumerRecord&lt;&gt;("my_topic", 0, 3L, "mykey", "myvalue3"));
      consumer.addRecord(new ConsumerRecord&lt;&gt;("my_topic", 0, 4L, "mykey", "myvalue4"));
  }
  </code></pre>
  
  <p>Since it's a <code>runnable</code> and consumer is not exposed this test not working as expected. How may I fix this?</p>