title: Spring Kafka multiple consumer for single topic consume different messages
body: <p>In my Spring Boot Kafka application, I have the following consumer configuration:</p>
  
  <pre><code>@Bean
  public ConsumerFactory&lt;String, Post&gt; postConsumerFactory(KafkaProperties kafkaProperties) {
      return new DefaultKafkaConsumerFactory&lt;&gt;(kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new JsonDeserializer&lt;&gt;(Post.class));
  }
  
  @Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Post&gt; postKafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
  
      kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerMaxPollIntervalMs);
      kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerMaxPollRecords);
  
      ConcurrentKafkaListenerContainerFactory&lt;String, Post&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
      factory.setConsumerFactory(postConsumerFactory(kafkaProperties));
  
      return factory;
  }
  </code></pre>
  
  <p>and the consumer:</p>
  
  <pre><code>@KafkaListener(topics = "${kafka.topic.post.send}", containerFactory = "postKafkaListenerContainerFactory")
  public void sendPost(ConsumerRecord&lt;String, Post&gt; consumerRecord, Acknowledgment ack) {
  
      // do some logic
  
      ack.acknowledge();
  }
  </code></pre>
  
  <p>If I understood correctly, right now I have one the single instance of my consumer.  I'd like to increase numbers of the post consumers, let's say to have 5 consumers that will consume different(not the same) messages from <code>${kafka.topic.post.send}</code> in order to speed up message consumption.</p>
  
  <p>Is it as simple as add <code>factory.setConcurrency(5);</code> to my <code>postKafkaListenerContainerFactory()</code>, for example:</p>
  
  <pre><code>@Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Post&gt; postKafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
  
      kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerMaxPollIntervalMs);
      kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerMaxPollRecords);
  
      ConcurrentKafkaListenerContainerFactory&lt;String, Post&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
      factory.setConsumerFactory(postConsumerFactory(kafkaProperties));
      factory.setConcurrency(5);
  
      return factory;
  }
  </code></pre>
  
  <p>or do I need to do some extra work in order to achieve it?</p>