title: How to avoid poll() retreiving same message if commitSync() is not called
body: <p>I'm a bit confused about poll() expected behavior. In my app, if the processLogic() works, then I should commit manually the offset so when the next poll() is called I do receive the new messages.</p>
  
  <p>Problem occurs when processLogic() throws an error. I set the consumer to seek to the offset failed during processing. On next poll(), it again receives the same message.(correct behavior as I ordered consumer to manually reset the offset to that position)Imagine it works fine and also doCommitSync() is called.</p>
  
  <p>The unexpected behavior occurs in the following poll(). It should take the new messages, but it still retrieve the last message, which leads to call again to the processLogic() function and doCommitSync(). It also throws the following error during doCommitSync(): </p>
  
  <blockquote>
    <p>Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.</p>
  </blockquote>
  
  <p>My consumer config:</p>
  
  <pre><code>enable.auto.commit=false
  
  isolation.level=read_committed
  
  auto.offset.reset=latest
  
  public void runConsumer() {
      Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
  
      try {
        consumer.subscribe(topics);
  
        while (!closed.get()) {
  
          processedStatus.set(false);
  
          final ConsumerRecords&lt;String, String&gt; consumedRecords = consumer.poll(numRecords);
          if (!consumedRecords.isEmpty()) {
            StreamSupport.stream(consumedRecords.spliterator(), false)
              .map(ConsumerRecord::value)
              .forEach(record -&gt; {
                try {
                  processLogic(); //do some logic which can fail
                  processedStatus.set(true);
                } catch (Exception e) {
                  logger.error("Error applying action: " + record.getUuid(), e);
                }
                if (processedStatus.get()) {
                  doCommitSync();
                } else {
                  consumer.seek(new TopicPartition(recordTopic, recordPartition), recordOffset);
                }
              });
          }
        }
      } catch (WakeupException e) {
        logger.error("Kafka Consumer wakeup exception");
      } finally {
        alertConsumer.close();
        shutdownLatch.countDown();
      }
  </code></pre>