title: Kafka consumer batch does not return as per max poll record config
body: <p>MY kafka Listener detail as below:</p>
  <pre><code>@KafkaListener(
      topics=&quot;test&quot;,
      groupId=&quot;groupid&quot;,
      containerFactory=&quot;kafkaBatchListenerContainerFactory&quot; )
   public void onMessage(List&lt;ConsumerRecord&lt;String,String&gt;&gt; messages, Acknowledgment acknowledgment) throws Exception {
      log.info(&quot;Batch size is {}&quot;,messages.size());
      acknowledgment.acknowledge();
  }
  </code></pre>
  <p>Bean Configuration detail:</p>
  <pre><code>@Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaBatchListenerContainerFactory() throws IOException {
      ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
      factory.setConcurrency(3); //Equal to partition
      factory.getContainerProperties().setAckMode(&quot;MANUAL&quot;);
      
      factory.getContainerProperties().setIdleBetweenPolls((long)  1 * 1000L);
      factory.getContainerProperties().setPollTimeout(kafkaConfigProperties.getPollTimeout() * 1000L);
      factory.setMissingTopicsFatal(false);
      factory.setBatchListener(true);
      factory.getContainerProperties().setAsyncAcks(true);
      factory.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1000L, 4)));
  }
  </code></pre>
  <p>Consumer Factory:</p>
  <pre><code>@Bean
  public ConsumerFactory&lt;String, String&gt; consumerFactory() throws IOException {
      Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
      config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;localhost:9080&quot;);
  
      config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
      config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
      config.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
      config.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
  
      config.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;groupid&quot;);
      config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
      config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
  
      config.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,  60 * 1000);
      config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 2 * 1000);
      config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30 * 1000);
      config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 900 * 1000);
  
      config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
      config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, read_committed);
  
      
  
      return new DefaultKafkaConsumerFactory&lt;&gt;(config, new StringDeserializer(), new StringDeserializer());
  }
  </code></pre>
  <p>When stop and start the server every time, able to receive batch of records. However during live transaction, received one record instead of batch. Let me know any suggestion.</p>
  <p>My logs as below :</p>
  <pre><code>Batch size is 1
  Batch size is 1
  Batch size is 1
  Batch size is 1
  </code></pre>