title: Spring-Kafka Concurrency Property
body: <p>I am progressing on writing my first Kafka Consumer by using Spring-Kafka. Had a look at the different options provided by framework, and have few doubts on the same. Can someone please clarify below if you have already worked on it. </p>
  
  <p><strong>Question - 1</strong> : As per Spring-Kafka documentation, there are 2 ways to implement Kafka-Consumer; "You can receive messages by configuring a MessageListenerContainer and providing a message listener or by using the @KafkaListener annotation". Can someone tell when should I choose one option over another ?</p>
  
  <p><strong>Question - 2</strong> : I have chosen KafkaListener approach for writing my application. For this I need to initialize a container factory instance and inside container factory there is option to control concurrency. Just want to double check if my understanding about concurrency is correct or not. </p>
  
  <p>Suppose, I have a topic name MyTopic which has 4 partitions in it. And to consume messages from MyTopic, I've started 2 instances of my application and these instances are started by setting concurrency as 2. So, Ideally as per kafka assignment strategy, 2 partitions should go to consumer1 and 2 other partitions should go to consumer2. Since the concurrency is set as 2, does each of the consumer will start 2 threads, and will consume data from the topics in parallel ? Also should we consider anything if we are consuming in parallel. </p>
  
  <p><strong>Question 3</strong> - I have chosen manual ack mode, and not managing the offsets externally (not persisting it to any database/filesystem). So should I need to write custom code to handle rebalance, or framework will manage it automatically ? I think no as I am acknowledging only after processing all the records. </p>
  
  <p><strong>Question - 4</strong> : Also, with Manual ACK mode, which Listener will give more performance? BATCH Message Listener or normal Message Listener. I guess if I use Normal Message listener, the offsets will be committed after processing each of the messages. </p>
  
  <p>Pasted the code below for your reference.  </p>
  
  <p><strong>Batch Acknowledgement Consumer</strong>: </p>
  
  <pre><code>    public void onMessage(List&lt;ConsumerRecord&lt;String, String&gt;&gt; records, Acknowledgment acknowledgment,
            Consumer&lt;?, ?&gt; consumer) {
        for (ConsumerRecord&lt;String, String&gt; record : records) {
            System.out.println("Record : " + record.value());
            // Process the message here..
            listener.addOffset(record.topic(), record.partition(), record.offset());
         }
         acknowledgment.acknowledge();
      }
  </code></pre>
  
  <p><strong>Initialising container factory:</strong></p>
  
  <pre><code>@Bean
  public ConsumerFactory&lt;String, String&gt; consumerFactory() {
      return new DefaultKafkaConsumerFactory&lt;String, String&gt;(consumerConfigs());
  }
  
  @Bean
  public Map&lt;String, Object&gt; consumerConfigs() {
      Map&lt;String, Object&gt; configs = new HashMap&lt;String, Object&gt;();
      configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer);
      configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
      configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enablAutoCommit);
      configs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPolInterval);
      configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
      configs.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
      configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      return configs;
  }
  
  @Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;String, String&gt;();
      // Not sure about the impact of this property, so going with 1
      factory.setConcurrency(2);
      factory.setBatchListener(true);
      factory.getContainerProperties().setAckMode(AckMode.MANUAL);
      factory.getContainerProperties().setConsumerRebalanceListener(RebalanceListener.getInstance());
      factory.setConsumerFactory(consumerFactory());
      factory.getContainerProperties().setMessageListener(new BatchAckConsumer());
      return factory;
  }
  </code></pre>