title: kafka async commit request failing
body: <p>I have observed that the kafka consumer lag suddenly starts increasing after few hours/days running.</p>
  
  <p>Upon checking the logs, I see alot of exception:</p>
  
  <blockquote>
    <p>org.apache.kafka.clients.consumer.RetriableCommitFailedException: Offset commit failed with a retriable exception. You should retry committing the latest consumed offsets.</p>
  </blockquote>
  
  <p>My ConsumerThread class:</p>
  
  <pre><code>public class ConsumerThread implements Runnable {
    private final KafkaConsumer&lt;String, Map&lt;String, Object&gt;&gt; consumer;
    public ConsumerThread(
      this.consumer = new KafkaConsumer&lt;&gt;(getConsumerConfig(kafkaConfiguration));
    }
  
    @Override
    public void run() {
      try {
        consumer.subscribe(topicList);
  
        while (true) {
          ConsumerRecords&lt;String, Map&lt;String, Object&gt;&gt; records =
              consumer.poll(Duration.ofMillis(kafkaConfiguration.getPollIntervalMs()));
  
          long startPerPoll = System.nanoTime();
          for (final ConsumerRecord&lt;String, Map&lt;String, Object&gt;&gt; record : records) {
              // message processing logic here
          }
  
  
          consumer.commitAsync((offsets, exception) -&gt; {
              if (exception != null) {
                //log.error(exception.getMessage());
                log.info("exception while committing offset, consumerThread: {}, exception: {}", Thread.currentThread().getName(), exception);
                exception.printStackTrace();
              }
          });
  
  
        }
      } catch (Exception e) {
        // ignore for shutdown
        log.info("exception in run for consumerThread: {}", e);
      } finally {
        try {
          if (Objects.nonNull(consumer)) {
            consumer.commitSync();
          }
        } finally {
          if (Objects.nonNull(consumer)) {
            consumer.close();
          }
        }
      }
  }
  </code></pre>
  
  <p>My kafka configuration:</p>
  
  <pre><code>groupId: cep-cg
  autoCommitEnabled: false
  sessionTimeoutMs: 30000
  heartBeatIntervalMs: 10000
  autoOffsetReset: latest
  maxPollRecord: 250
  maxPollIntervalMs: 180000
  requestTimeoutMs: 240000
  pollIntervalMs: 3000
  </code></pre>
  
  <p>I checked other answers on stackoverflow and made few tweaks but none of them seems to work.</p>
  
  <p>What I want to know is:</p>
  
  <ol>
  <li><p>Any leads on why the lag could be increasing suddenly ?</p></li>
  <li><p>Is it possible alot of commitAsync request is pending on the broker and maybe some time (defined by some config on broker), the commitAsync request starts to fail ?</p></li>
  <li><p>Let's say a consumerThread took more that <code>max.poll.interval.ms</code> time to process the message. In this case, it would be kicked out of the group and the re-balance is triggered. Now all the <code>commitAsync</code> request pending on broker fails with <code>CommitFailedException</code> since the partition now belongs to some other consumer in group.  In the above code, the consumer would break out of infinite loop and will be closed forever. Is this correct way ? Or should I catch <code>CommitFailedException</code> and again resume the loop to keep the consumer alive ?</p></li>
  </ol>