title: Spring @KafkaListener and concurrency
body: <p>I am working with spring boot + spring @KafkaListener. And the behavior I expect is: my kafka listener reads messages in 10 threads. So that, if one of threads hangs, other messages are would continue reading and handling messages.</p>
  
  <p>I defined bean of</p>
  
  <pre><code>@Bean
  public ConcurrentKafkaListenerContainerFactory&lt;?, ?&gt; kafkaListenerContainerFactory(
          ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
          ConsumerFactory&lt;Object, Object&gt; kafkaConsumerFactory)
  {
  
      ConcurrentKafkaListenerContainerFactory&lt;Object, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      configurer.configure(factory, kafkaConsumerFactory);
      factory.getContainerProperties().setMissingTopicsFatal(false);
      factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.INFO);
      return factory;
  }
  </code></pre>
  
  <p>And spring boot config:</p>
  
  <pre><code>spring.kafka.listener.concurrency=10
  </code></pre>
  
  <p>I see that all configs work, I see my 10 threads in jmx:</p>
  
  <p><a href="https://i.stack.imgur.com/IdY4l.png" rel="noreferrer"><img src="https://i.stack.imgur.com/IdY4l.png" alt="enter image description here"></a></p>
  
  <p>But then I make such test:</p>
  
  <pre><code> @KafkaListener(topics = {
              "${topic.name}" }, clientIdPrefix = "${kafka.client.id.prefix}", idIsGroup = false, id = "${kafka.listener.name}", containerFactory = "kafkaListenerContainerFactory")
      public void listen(ConsumerRecord&lt;String, String&gt; record)
      {
          if(record.getVersion() &lt; 3) {
              try {
                  Thread.sleep(20000);
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
          }
          else
              System.out.println("It works!");
  
      }
  </code></pre>
  
  <p>If version is &lt; 3, then hang, otherwise - work.
  I send 3 messages with version 1,2 and 3. I expect that messages with version 1 and 2 will hang, but version 3 will be processed at the time it comes to listener. But unfortunately message with version 3 waits for messages 1 and 2 before starts its processing.</p>
  
  <p>Maybe my expectations are not true and this is a right behavior of kafka listener. 
  Please help me to deal with kafka concurrency, why does it act like that?</p>