title: Wait until @KafkaListener finishes processing message in tests with @EmbeddedKafka
body: <p>I have a <code>@KafkaListener</code> consumer and want to write integration test.
  The fact is that it seems to be difficult to find the exact moment when method <code>Consumer#consume</code> finished its execution to perform some asserts after message was processed and some state in database has changed.</p>
  
  <pre><code>@Component
  public class Consumer {
  
      private final Service service;
  
      @KafkaListener(id = "id", groupId = "group", topics = "topic", containerFactory = "factory")
      public void consume(@Payload Message message, Acknowledgment acknowledgment) {
          service.process(message);
          acknowledgment.acknowledge();
      }
  
  }
  </code></pre>
  
  <p>Test</p>
  
  <pre><code>@SpringBootTest
  @EmbeddedKafka
  void class Testing {
      // some useful beans 
  
      @SpyBean
      private Consumer consumer;
  
      @Test
      void shoudConsume() throws Exception {
          Message message = new Message();
          String topic = "topic";
          Map&lt;String, Object&gt; senderProps = KafkaTestUtils.producerProps(embeddedKafka);
          new KafkaProducer&lt;String, String&gt;(senderProps).send(new ProducerRecord&lt;&gt;(topic, message))
                  .get(1L, TimeUnit.SECONDS);
  
          Mockito.verify(consumer, Mockito.timeout(1_000L)).consume(any(Message.class), any(Acknowledgment.class));
          // perform some asserts
      }
  </code></pre>
  
  <p>The fact is that if I put <code>Thread.sleep(1000L)</code> the consumer processes message and all works fine but with Mockito it doesn't work, because all asserts executes before consumer finished its execution of method <code>Consumer#consume</code>.</p>
  
  <p>Is there is a opportunity (using Listeners, or etc) to catch the moment when <code>@KafkaListener</code>-consumer acknowledged/finished message processing to perform asserts with appropriate database state? The integration test is needed to be sure that end-to-end functionality works fine.</p>
  
  <p>Also I tried to make <code>#verify</code> checks on <code>@SpyBean private Service service</code>, method <code>Service#process</code>, but it doesn't work too.</p>