title: Unit Test Kafka Consumer Written Using Spring Kafka
body: <p>I have created a Kakfa consumer application (using spring kafka) and it seems to be working fine. Now, I am trying to add unit test cases for the same. 
  My consumer is batch acknowledgement consumer and it is started inside a container. The details of the consumer can be found in a below stack overflow post (which is asked by me)</p>
  
  <p><a href="https://stackoverflow.com/questions/55627430/spring-kafka-concurrency-property">Spring-Kafka Concurrency Property</a></p>
  
  <p>I am able to write a test case for my consumer by doing some research, but here I had to replicate the creation of container and few other things in order to make it work. I was wondering, 
  if there is any other way for test cases to use the same container which is started during application startup (may by pointing test context) and the consumer which is started during startup to use the embedded kafka instance directly. </p>
  
  <p>The solution I figured out is as below, but not sure this is the right approach or not. </p>
  
  <pre><code>@RunWith(SpringRunner.class)
  @SpringBootTest(classes = Launcher.class)
  public class BatchConsumerTest {
  
      @ClassRule
      public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "topic1");
      @Autowired
      private KafkaListenerEndpointRegistry registry;
  
      @Autowired
      private KafkaTemplate&lt;String, String&gt; template;
  
      @MockBean
      private RestService restService;
  
      @Before
      public void setup() {
          Mockito.when(restService.invokeService("")).thenReturn("");
      }
  
      @Test
      public void test() throws Exception {
          ConcurrentMessageListenerContainer&lt;?, ?&gt; container = null;
          for (MessageListenerContainer con : registry.getAllListenerContainers()) {
              container = (ConcurrentMessageListenerContainer&lt;?, ?&gt;) con;
              container.stop();
          }
          Map&lt;String, Object&gt; consumerProps = KafkaTestUtils.consumerProps("group1", "false",
                  embeddedKafka.getEmbeddedKafka());
          DefaultKafkaConsumerFactory&lt;String, String&gt; factory = new DefaultKafkaConsumerFactory&lt;String, String&gt;(
                  consumerProps);
          ContainerProperties containerProps = new ContainerProperties("ibo-grocerybag-subscription");
          containerProps.setAckMode(AckMode.MANUAL);
          ConcurrentMessageListenerContainer&lt;String, String&gt; messageListContainer = new ConcurrentMessageListenerContainer&lt;String, String&gt;(
                  factory, containerProps);
          BatchAcknowledgingConsumerAwareMessageListener&lt;String, String&gt; listener = new BatchConsumer();
          CountDownLatch latch = new CountDownLatch(1);
          // messageListContainer.setupMessageListener(listener);
          messageListContainer.setupMessageListener(new BatchAcknowledgingConsumerAwareMessageListener&lt;String, String&gt;() {
  
              @Override
              public void onMessage(List&lt;ConsumerRecord&lt;String, String&gt;&gt; data, Acknowledgment acknowledgment,
                      Consumer&lt;?, ?&gt; consumer) {
                  System.out.println("*******Data : " + data.get(0).value());
                  listener.onMessage(data, acknowledgment, consumer);
                  latch.countDown();
              }
          });
          messageListContainer.start();
          ContainerTestUtils.waitForAssignment(messageListContainer,
                  embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic());
          template.send("topic1", "Hello");
          latch.await(10000, TimeUnit.MILLISECONDS);
          assertThat(latch.getCount()).isEqualTo(0);
      }
  
      @After
      public void destroy() {
          embeddedKafka.getEmbeddedKafka().destroy();
      }
  </code></pre>