[CAMEL-17489] camel-kafka - Unsubscribing fails due to already closed consumer

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 3.14.0
• Fix Version/s: 3.14.1, 3.15.0
• Component/s: camel-kafka
• Estimated Complexity: Unknown
• Assignee: Unassigned
• Reporter: Rafał Gała
• Created: 14/Jan/22 11:15
• Updated: 24/Jan/22 05:56
• Resolved: 24/Jan/22 05:56

Description

In KafkaFetchRecords, when an exception occurs inside startPolling method, the consumer is closed in finally block:

finally {
    lock.unlock();

    // only close if not retry
    if (!isRetrying()) {
        LOG.debug("Closing consumer {}", threadId);
        IOHelper.close(consumer);
    }
} 

 and then unsubscribing in run method fails with "This consumer has already been closed"

LOG.info("Terminating KafkaConsumer thread: {} receiving from topic: {}", threadId, topicName);
safeUnsubscribe();
IOHelper.close(consumer); 

Issue Links

• relates to: CAMEL-17493 camel-kafka: safe unsubscription should ignore safe exceptions
• links to: GitHub Pull Request #6751