[KAFKA-3491] Issue with consumer close() in finally block with 'enable.auto.commit=true'

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 0.9.0.0, 0.9.0.1
• Component/s: consumer
• Assignee: Jason Gustafson
• Reporter: dan norwood
• Created: 01/Apr/16 17:36
• Updated: 05/Mar/17 21:39

Description

imagine you have a run loop that looks like the following:

  public void run() {
    try {
      consumer.subscribe(topics);

      while (true) {
        ConsumerRecords<K, V> records = consumer.poll(Long.MAX_VALUE);
        records.forEach(record -> process(record));
      }
    } catch (WakeupException e) {
      // ignore, we're closing
    } catch (Exception e) {
      log.error("Unexpected error", e);
    } finally {
      consumer.close();
    }
  }

if you run this with 'enable.auto.commit=true' and throw an exception in the 'process()' method you will still try to commit all the read, but unprocessed, offsets in the most recent batch.

Issue Links

• blocks: KAFKA-4242 Re-enable committing offsets in consumer's unsubscribe() when auto commit is enabled
• relates to: KAFKA-4234 Consumer should not commit offsets in unsubscribe()