[camel] branch main updated: CAMEL-17489: camel-kafka - Unsubscribe before closing the consumer

[orpiske]

This is an automated email from the ASF dual-hosted git repository.

orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 3a70856  CAMEL-17489: camel-kafka - Unsubscribe before closing the 
consumer
3a70856 is described below

commit 3a70856fc34bdd245f71bd456457ceb03e238b2d
Author: Rafał Gała <rafal.g...@ing.pl>
AuthorDate: Fri Jan 14 12:22:29 2022 +0100

    CAMEL-17489: camel-kafka - Unsubscribe before closing the consumer
---
 .../main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java    | 1 +
 1 file changed, 1 insertion(+)

diff --git 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
index ccf2119..d51af7a 100644
--- 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
+++ 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
@@ -212,6 +212,7 @@ class KafkaFetchRecords implements Runnable {
             // only close if not retry
             if (!isRetrying()) {
                 LOG.debug("Closing consumer {}", threadId);
+                safeUnsubscribe();
                 IOHelper.close(consumer);
             }
         }