title: How to write Unit test for @KafkaListener?
body: <p>Trying to figure out if I can write unit test for @KafkaListener using spring-kafka and spring-kafka-test. </p>
  
  <p>My Listener class. </p>
  
  <pre class="lang-java prettyprint-override"><code>    public class MyKafkaListener {
      @Autowired
      private MyMessageProcessor myMessageProcessor;
  
      @KafkaListener(topics = "${kafka.topic.01}", groupId = "SF.CLIENT", clientIdPrefix = "SF.01", containerFactory = "myMessageListenerContainerFactory")
      public void myMessageListener(MyMessage message) {
          myMessageProcessor.process(message);
          log.info("MyMessage processed");
      }}
  </code></pre>
  
  <p>My Test class :</p>
  
  <pre class="lang-java prettyprint-override"><code>    @RunWith(SpringRunner.class)
      @DirtiesContext
      @EmbeddedKafka(partitions = 1, topics = {"I1.Topic.json.001"})
      @ContextConfiguration(classes = {TestKafkaConfig.class})
      public class MyMessageConsumersTest {
  
      @Autowired
      private MyMessageProcessor myMessageProcessor;
  
      @Value("${kafka.topic.01}")
      private String TOPIC_01;
  
      @Autowired
      private KafkaTemplate&lt;String, MyMessage&gt; messageProducer;
  
      @Test
      public void testSalesforceMessageListner() {
          MyMessageConsumers myMessageConsumers = new MyMessageConsumers(mockService);
          messageProducer.send(TOPIC_01, "MessageID", new MyMessage());
          verify(myMessageProcessor, times(1)).process(any(MyMessage.class));
      }}
  </code></pre>
  
  <p>My Test config class :</p>
  
  <pre class="lang-java prettyprint-override"><code>    @Configuration
      @EnableKafka
      public class TestKafkaConfig {
      @Bean
      public MyMessageProcessor myMessageProcessor() {
          return mock(MyMessageProcessor.class);
      }
      @Bean
      public KafkaEmbedded kafkaEmbedded() {
          return new KafkaEmbedded(1, true, 1, "I1.Topic.json.001");
      }
  
      //Consumer
      @Bean
      public ConsumerFactory&lt;String, MyMessage&gt; myMessageConsumerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded().getBrokersAsString());
          props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          return new DefaultKafkaConsumerFactory&lt;&gt;(props, new StringDeserializer(), new JsonDeserializer&lt;&gt;(MyMessage.class));
      }
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, MyMessage&gt; myMessageListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, MyMessage&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(myMessageConsumerFactory());
          return factory;
      }
  
      //Producer
      @Bean
      public ProducerFactory&lt;String, MyMessage&gt; producerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded().getBrokersAsString());
          props.put(ProducerConfig.RETRIES_CONFIG, 0);
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaMessageSerializer.class);
          return new DefaultKafkaProducerFactory&lt;&gt;(props);
      }
      @Bean
      public KafkaTemplate&lt;String, MyMessage&gt; messageProducer() {
          return new KafkaTemplate&lt;&gt;(producerFactory());
      }
      }
  
  </code></pre>
  
  <p>Is there any simple way to make this work ?</p>
  
  <p>Or should I do the testing of @KafkaListener in some other way ? In unit test, how do I ensure @KafkaListener is invoked when a new message is arrived in Kafka.</p>