Clean cached messages on destroy kafka consumer

azat · azat · commit 853fdfe77510 · 2023-12-29T14:30:21.000+01:00
The callchain of the kafka consumer is very tricky, so for the sake of common sense let's just clean the messages on moving out consumer (and in dtor, but this is just to keep that two code path in sync). (Also reported by @filimonov) Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>
diff --git a/src/Storages/Kafka/KafkaConsumer.cpp b/src/Storages/Kafka/KafkaConsumer.cpp
@@ -153,6 +153,7 @@ void KafkaConsumer::createConsumer(cppkafka::Configuration consumer_config)
 
 ConsumerPtr && KafkaConsumer::moveConsumer()
 {
+    cleanUnprocessed();
     if (!consumer->get_subscription().empty())
     {
         try
@@ -173,6 +174,7 @@ KafkaConsumer::~KafkaConsumer()
     if (!consumer)
         return;
 
+    cleanUnprocessed();
     try
     {
         if (!consumer->get_subscription().empty())

Original file line number	Diff line number	Diff line change
`@@ -153,6 +153,7 @@ void KafkaConsumer::createConsumer(cppkafka::Configuration consumer_config)`
`153`	`153`
`154`	`154`	`ConsumerPtr && KafkaConsumer::moveConsumer()`
`155`	`155`	`{`
	`156`	`+ cleanUnprocessed();`
`156`	`157`	`if (!consumer->get_subscription().empty())`
`157`	`158`	`{`
`158`	`159`	`try`
`@@ -173,6 +174,7 @@ KafkaConsumer::~KafkaConsumer()`
`173`	`174`	`if (!consumer)`
`174`	`175`	`return;`
`175`	`176`
	`177`	`+ cleanUnprocessed();`
`176`	`178`	`try`
`177`	`179`	`{`
`178`	`180`	`if (!consumer->get_subscription().empty())`