title: Listener not consuming message on Test with EmbeddedKafka Spring
body: <p>I was trying to see if a service is invoked when the consumer receives a message from kafka topic but the test is not passing and the consumer is not even receiving the message.</p>
  <p>My test:</p>
  <pre><code>@SpringBootTest
  @DirtiesContext
  @EmbeddedKafka(partitions = 1, brokerProperties = { &quot;listeners=PLAINTEXT://localhost:9092&quot;, &quot;port=9092&quot; })
  class ConsumerTest {
  
      @Autowired
      Consumer consumer;
  
      @Mock
      private Service service;
  
      @Autowired
      private KafkaTemplate&lt;String, String&gt; kafkaTemplate;
      
      @Test
      public void givenEmbeddedKafkaBroker_whenSendingtoSimpleProducer_thenMessageReceived() 
        throws Exception {
  
          String message = &quot;hi&quot;;
  
          kafkaTemplate.send(&quot;topic&quot;, message);
  
          consumer.getLatch().await(10000, TimeUnit.MILLISECONDS);
  
         
          verify(service, times(1)).addMessage();
      }
  }
  
  </code></pre>
  <p>The consumer, in main package, is a normal consumer with @KafkaListener(topics = &quot;topic&quot;).
  Then I have a configuration file:</p>
  <pre><code>@EnableKafka
  @Configuration
  public class KafkaConfig {
    
    @Bean
    public Map&lt;String, Object&gt; consumerConfigs() {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;localhost:9092&quot;);
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;group&quot;);
      return props;
    }
  
    @Bean
    public ConsumerFactory&lt;String, String&gt; consumerFactory() {
      return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs(), new StringDeserializer(),
          new StringDeserializer());
    }
  
    @Bean
    public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory =
          new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
  
      return factory;
    }
  
  
  }
  </code></pre>
  <p>And also in application.properties (inside test package) i put this:</p>
  <p>spring:
  kafka:
  consumer:
  auto-offset-reset: earliest
  group-id: group</p>