title: Kafka Consumer not able to read all message after offset commit (error=OFFSET_OUT_OF_RANGE
body: <p>i created consumer to receive messages as batch ,
  ConsumerConfig:</p>
  <pre><code>    allow.auto.create.topics = false
      auto.commit.interval.ms = 5000
      auto.offset.reset = latest
      bootstrap.servers = [localhost:9092]
      check.crcs = true
      client.dns.lookup = default
      client.id =
      client.rack =
      connections.max.idle.ms = 540000
      default.api.timeout.ms = 60000
      enable.auto.commit = false
      exclude.internal.topics = true
      fetch.max.bytes = 52428800
      fetch.max.wait.ms = 500
      fetch.min.bytes = 1
      group.id = cm-persistence-notification
      group.instance.id = null
      heartbeat.interval.ms = 3000
      interceptor.classes = []
      internal.leave.group.on.close = true
      isolation.level = read_uncommitted
      key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
      max.partition.fetch.bytes = 1048576
      max.poll.interval.ms = 300000
      max.poll.records = 1000
      metadata.max.age.ms = 300000
      metric.reporters = []
      metrics.num.samples = 2
      metrics.recording.level = INFO
      metrics.sample.window.ms = 30000
      partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
      receive.buffer.bytes = 65536
      reconnect.backoff.max.ms = 1000
      reconnect.backoff.ms = 50
      request.timeout.ms = 30000
      retry.backoff.ms = 100
      sasl.client.callback.handler.class = null
      sasl.jaas.config = null
      sasl.kerberos.kinit.cmd = /usr/bin/kinit
      sasl.kerberos.min.time.before.relogin = 60000
      sasl.kerberos.service.name = null
      sasl.kerberos.ticket.renew.jitter = 0.05
      sasl.kerberos.ticket.renew.window.factor = 0.8
      sasl.login.callback.handler.class = null
      sasl.login.class = null
      sasl.login.refresh.buffer.seconds = 300
      sasl.login.refresh.min.period.seconds = 60
      sasl.login.refresh.window.factor = 0.8
      sasl.login.refresh.window.jitter = 0.05
      sasl.mechanism = GSSAPI
      security.protocol = PLAINTEXT
      send.buffer.bytes = 131072
      session.timeout.ms = 10000
      ssl.cipher.suites = null
      ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
      ssl.endpoint.identification.algorithm = https
      ssl.key.password = null
      ssl.keymanager.algorithm = SunX509
      ssl.keystore.location = null
      ssl.keystore.password = null
      ssl.keystore.type = JKS
      ssl.protocol = TLS
      ssl.provider = null
      ssl.secure.random.implementation = null
      ssl.trustmanager.algorithm = PKIX
      ssl.truststore.location = null
      ssl.truststore.password = null
      ssl.truststore.type = JKS
      value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  </code></pre>
  <p>Spring boot Config:</p>
  <pre><code>public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, String&gt;&gt; cmPersistenceListenerContainerFactory(
          KafkaProperties kafkaProperties )
      {
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; containerFactory =
              new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
  
          Map&lt;String, Object&gt; consumerProperties = kafkaProperties.buildConsumerProperties();
          consumerProperties.put( ConsumerConfig.MAX_POLL_RECORDS_CONFIG, &quot;1000&quot; );
          consumerProperties.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false );
          consumerProperties.put( ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false );
  
          containerFactory
              .setConsumerFactory(
                  new DefaultKafkaConsumerFactory&lt;&gt;(
                      consumerProperties, new StringDeserializer(), new StringDeserializer() ) );
          containerFactory.setBatchListener( true );
         
          containerFactory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.INFO);
          
          containerFactory.getContainerProperties().setAckMode( AckMode.MANUAL_IMMEDIATE );
          
          return containerFactory;
      }
  
  
      @Bean
      public KafkaAdmin kafkaAdmin( KafkaProperties kafkaProperties )
      {
          return new KafkaAdmin( kafkaProperties.buildAdminProperties() );
      }
  </code></pre>
  <p>Listener Class :</p>
  <pre><code>@KafkaListener( id = &quot;batch-listener-0&quot;, topics = &quot;topic1&quot;, groupId = &quot;test&quot;, containerFactory = KafkaConsumerConfiguration.CONTAINER_FACTORY_NAME )
      public void receive(
          @Payload List&lt;String&gt; messages,
          @Header( KafkaHeaders.RECEIVED_MESSAGE_KEY ) List&lt;String&gt; keys,
          @Header( KafkaHeaders.RECEIVED_PARTITION_ID ) List&lt;Integer&gt; partitions,
          @Header( KafkaHeaders.RECEIVED_TOPIC ) List&lt;String&gt; topics,
          @Header( KafkaHeaders.OFFSET ) List&lt;Long&gt; offsets,
          Acknowledgment ack )
      {
          long startTime = System.currentTimeMillis();
  
          handleNotifications( messages ); // will take more than 5s to process all messages
  
          long endTime = System.currentTimeMillis();
  
          long timeElapsed = endTime - startTime;
  
          LOGGER.info( &quot;Execution Time :{}&quot;, timeElapsed );
  
          ack.acknowledge();
  
          
          LOGGER.info( &quot;Acknowledgment Success&quot; );
  
      }
  </code></pre>
  <p>I am using manual Acknowledgment after processing message .</p>
  <p>I found some debug log :</p>
  <p><a href="https://i.stack.imgur.com/HIH9w.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/HIH9w.png" alt="enter image description here" /></a></p>
  <p>In above debug log , ****fetch offset is happening before offset commit, that offset is not commited so it returning OFFSET_OUT_OF_RANGE, consumer is not able to receive any message after that , Is there any way to handle this error in consumer code or how to fetch offset only after commit ****</p>