Unsubscribe kafka consumer before cleaning it by TTL

Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>

src/Storages/Kafka/KafkaConsumer.cpp

@@ -151,6 +151,23 @@ void KafkaConsumer::createConsumer(cppkafka::Configuration consumer_config)
     });
 }
 
+ConsumerPtr && KafkaConsumer::moveConsumer()
+{
+    if (!consumer->get_subscription().empty())
+    {
+        try
+        {
+            consumer->unsubscribe();
+        }
+        catch (const cppkafka::HandleException & e)
+        {
+            LOG_ERROR(log, "Error during unsubscribe: {}", e.what());
+        }
+        drain();
+    }
+    return std::move(consumer);
+}
+
 KafkaConsumer::~KafkaConsumer()
 {
     if (!consumer)

src/Storages/Kafka/KafkaConsumer.h

@@ -74,7 +74,7 @@ public:
 
     void createConsumer(cppkafka::Configuration consumer_config);
     bool hasConsumer() const { return consumer.get() != nullptr; }
-    ConsumerPtr && moveConsumer() { return std::move(consumer); }
+    ConsumerPtr && moveConsumer();
 
     void commit(); // Commit all processed messages.
     void subscribe(); // Subscribe internal consumer to topics.