title: Why is my kafka listener not working in my Unit Test
body: <p><strong>PROBLEM</strong></p>
  
  <p>I am attempting to unit test a kafka producer that sends an Integer. When running my unit test I can see outputs suggesting that my producer and consumer are working correctly in the console. That is the producer is sending a value of zero, the consumer received zero, and totaled that integer with the running total.</p>
  
  <pre><code>sending data = 0
  received content = 0
  sending total = 0
  </code></pre>
  
  <p>However the unit test ultimately fails when the test ConsumerRecord comes back with a null entry. I guess the listener is not working at all.</p>
  
  <pre><code>java.lang.AssertionError: 
  Expected: a ConsumerRecord with value 0
       but: is null
  </code></pre>
  
  <p><strong>QUESTION</strong></p>
  
  <p>Is there an error in the way I have defined the container/messageListener? Or something more fundamentally wrong with my unit test? </p>
  
  <p>This url contains the code of both my producer/config and consumer/config. It might be easier to view it there rather than making code section huge on here.</p>
  
  <p><a href="https://github.com/ewingian/RestCalculator/tree/master/src/main/java/com/calculator/kafka" rel="nofollow noreferrer">https://github.com/ewingian/RestCalculator/tree/master/src/main/java/com/calculator/kafka</a></p>
  
  <p><strong>Unit Test</strong></p>
  
  <pre><code>package com.calculator;
  
  /**
   * Created by ian on 2/9/18.
   */
  import com.calculator.kafka.services.KafkaProducer;
  import com.calculator.kafka.services.KafkaConsumer;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.ClassRule;
  import org.junit.Test;
  import org.junit.runner.RunWith;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.boot.test.context.SpringBootTest;
  import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  import org.springframework.kafka.core.KafkaTemplate;
  import org.springframework.kafka.core.ProducerFactory;
  import org.springframework.kafka.listener.KafkaMessageListenerContainer;
  import org.springframework.kafka.listener.MessageListener;
  import org.springframework.kafka.listener.config.ContainerProperties;
  import org.springframework.kafka.test.rule.KafkaEmbedded;
  import org.springframework.kafka.test.utils.ContainerTestUtils;
  import org.springframework.kafka.test.utils.KafkaTestUtils;
  import org.springframework.test.annotation.DirtiesContext;
  import org.springframework.test.context.junit4.SpringRunner;
  
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  
  import static org.junit.Assert.assertThat;
  import static org.springframework.kafka.test.assertj.KafkaConditions.key;
  import static org.springframework.kafka.test.hamcrest.KafkaMatchers.*;
  
  @RunWith(SpringRunner.class)
  @SpringBootTest
  @DirtiesContext
  public class KafkaTest {
  
      // in case I need to send some integers
      private Integer i1 = 0;
      private Integer i2 = 3;
  
      private static final String SENDER_TOPIC = "input";
  
      private List&lt;Integer&gt; l1;
  
      @Autowired
      private KafkaProducer producer;
  
      @Autowired
      private KafkaConsumer consumer;
  
      private KafkaMessageListenerContainer&lt;String, Integer&gt; container;
  
      private BlockingQueue&lt;ConsumerRecord&lt;String, Integer&gt;&gt; records;
  
      private static final Logger LOGGER = LoggerFactory.getLogger(KafkaTest.class);
  
  
      @ClassRule
      public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, SENDER_TOPIC);
  
  
  
      @Before
      public void testTemplate() throws Exception {
  
          // set up the Kafka consumer properties
          Map&lt;String, Object&gt; consumerProperties = KafkaTestUtils.consumerProps("test-group", "false", embeddedKafka);
  
          // create a Kafka consumer factory
          DefaultKafkaConsumerFactory&lt;String, Integer&gt; consumerFactory = new DefaultKafkaConsumerFactory&lt;&gt;(consumerProperties);
  
          // set the topic that needs to be consumed
          ContainerProperties containerProperties = new ContainerProperties(SENDER_TOPIC);
  
          // create a Kafka MessageListenerContainer
          container = new KafkaMessageListenerContainer&lt;&gt;(consumerFactory, containerProperties);
  
          // create a thread safe queue to store the received message
          records = new LinkedBlockingQueue&lt;&gt;();
  
          // setup a Kafka message listener
          container.setupMessageListener(new MessageListener&lt;String, Integer&gt;() {
              @Override
              public void onMessage(ConsumerRecord&lt;String, Integer&gt; record) {
                  LOGGER.debug("test-listener received message='{}'", record.toString());
                  records.add(record);
              }
          });
  
          // start the container and underlying message listener
          container.start();
  
          // wait until the container has the required number of assigned partitions
          ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
  
  
      }
  
      @After
      public void tearDown() {
          // stop the container
          container.stop();
      }
  
      @Test
      public void testSend() throws InterruptedException {
          // send the message
          producer.send(i1);
  
          // check that the message was received
          ConsumerRecord&lt;String, Integer&gt; received = records.poll(10, TimeUnit.SECONDS);
          // Hamcrest Matchers to check the value
          assertThat(received, hasValue(i1));
  
          // AssertJ Condition to check the key
          assertThat(received, hasKey(null));
      }
  }
  </code></pre>