Backport #80795 to 25.4: Properly wait consumers before shutting down Kafka engine

robot-clickhouse · robot-clickhouse · commit dcaa0a666c5c · 2025-05-27T15:12:31.000Z
diff --git a/src/Storages/Kafka/StorageKafka.cpp b/src/Storages/Kafka/StorageKafka.cpp
@@ -220,7 +220,7 @@ StorageKafka::StorageKafka(
         const auto & table = getStorageID().getTableName();
         const auto & thread_name = std::string("KfkCln:") + table;
         setThreadName(thread_name.c_str(), /*truncate=*/ true);
-        cleanConsumers();
+        cleanConsumersByTTL();
     });
 }
 
@@ -318,11 +318,10 @@ void StorageKafka::shutdown(bool)
     }
 
     {
-        std::lock_guard lock(mutex);
         LOG_TRACE(log, "Closing {} consumers", consumers.size());
         Stopwatch watch;
-        consumers.clear();
-        LOG_TRACE(log, "Consumers closed. Took {} ms.", watch.elapsedMilliseconds());
+        cleanConsumers();
+        LOG_TRACE(log, "Consumers closed in {} ms.", watch.elapsedMilliseconds());
     }
 
     {
@@ -333,6 +332,40 @@ void StorageKafka::shutdown(bool)
     }
 }
 
+void StorageKafka::cleanConsumers()
+{
+    /// We need to clear the cppkafka::Consumer separately from KafkaConsumer, since cppkafka::Consumer holds a weak_ptr to the KafkaConsumer (for logging callback)
+    /// So if we will remove cppkafka::Consumer from KafkaConsumer destructor, then due to librdkafka will call the logging again from destructor, it will lead to a deadlock
+    std::vector<ConsumerPtr> consumers_to_close;
+
+    {
+        std::unique_lock lock(mutex);
+        /// Wait until all consumers will be released
+        cv.wait(lock, [&, this]()
+        {
+            auto it = std::find_if(consumers.begin(), consumers.end(), [](const auto & ptr)
+            {
+                return ptr->isInUse();
+            });
+            return it == consumers.end();
+        });
+
+        for (const auto & consumer : consumers)
+        {
+            if (!consumer->hasConsumer())
+                continue;
+            consumers_to_close.push_back(consumer->moveConsumer());
+        }
+    }
+
+    /// First close cppkafka::Consumer (it can use KafkaConsumer object via stat callback)
+    consumers_to_close.clear();
+
+    {
+        std::unique_lock lock(mutex);
+        consumers.clear();
+    }
+}
 
 void StorageKafka::pushConsumer(KafkaConsumerPtr consumer)
 {
@@ -342,17 +375,13 @@ void StorageKafka::pushConsumer(KafkaConsumerPtr consumer)
     CurrentMetrics::sub(CurrentMetrics::KafkaConsumersInUse, 1);
 }
 
-
-KafkaConsumerPtr StorageKafka::popConsumer()
-{
-    return popConsumer(std::chrono::milliseconds::zero());
-}
-
-
 KafkaConsumerPtr StorageKafka::popConsumer(std::chrono::milliseconds timeout)
 {
     std::unique_lock lock(mutex);
 
+    if (shutdown_called)
+        throw Exception(ErrorCodes::ABORTED, "Table is detached");
+
     KafkaConsumerPtr ret_consumer_ptr;
     std::optional<size_t> closed_consumer_index;
     for (size_t i = 0; i < consumers.size(); ++i)
@@ -457,7 +486,7 @@ cppkafka::Configuration StorageKafka::getProducerConfiguration()
     return KafkaConfigLoader::getProducerConfiguration(*this, params);
 }
 
-void StorageKafka::cleanConsumers()
+void StorageKafka::cleanConsumersByTTL()
 {
     UInt64 ttl_usec = (*kafka_settings)[KafkaSetting::kafka_consumers_pool_ttl_ms] * 1'000;
 
diff --git a/src/Storages/Kafka/StorageKafka.h b/src/Storages/Kafka/StorageKafka.h
@@ -77,7 +77,6 @@ class StorageKafka final : public IStorage, WithContext
     bool prefersLargeBlocks() const override { return false; }
 
     void pushConsumer(KafkaConsumerPtr consumer);
-    KafkaConsumerPtr popConsumer();
     KafkaConsumerPtr popConsumer(std::chrono::milliseconds timeout);
 
     const auto & getFormatName() const { return format_name; }
@@ -164,6 +163,7 @@ class StorageKafka final : public IStorage, WithContext
 
     bool streamToViews();
 
+    void cleanConsumersByTTL();
     void cleanConsumers();
 };
 
diff --git a/tests/queries/0_stateless/03522_storage_kafka_shutdown_smoke.reference b/tests/queries/0_stateless/03522_storage_kafka_shutdown_smoke.reference
diff --git a/tests/queries/0_stateless/03522_storage_kafka_shutdown_smoke.sql b/tests/queries/0_stateless/03522_storage_kafka_shutdown_smoke.sql
@@ -0,0 +1,24 @@
+-- Tags: no-fasttest
+-- Tag no-fasttest -- requires Kafka
+
+-- Regression test for proper StorageKafka shutdown
+-- https://github.com/ClickHouse/ClickHouse/issues/80674
+
+-- librdkafka may print some errors/warnings:
+---
+-- <Warning> StorageKafka (test_8g8g0dlf.kafka_test): sasl.kerberos.kinit.cmd configuration parameter is ignored.
+-- <Error> StorageKafka (test_8g8g0dlf.kafka_test): [client.id:ClickHouse-localhost-test_8g8g0dlf-kafka_test] [rdk:ERROR] [thrd:app]: ClickHouse-localhost-test_8g8g0dlf-kafka_test#consumer-1: 0.0.0.0:9092/bootstrap: Connect to ipv4#0.0.0.0:9092 failed: Connection refused (after 0ms in state CONNECT)
+-- <Error> StorageKafka (test_8g8g0dlf.kafka_test): [client.id:ClickHouse-localhost-test_8g8g0dlf-kafka_test] [rdk:ERROR] [thrd:app]: ClickHouse-localhost-test_8g8g0dlf-kafka_test#consumer-1: 0.0.0.0:9092/bootstrap: Connect to ipv4#0.0.0.0:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
+-- <Warning> StorageKafka (test_8g8g0dlf.kafka_test): Can't get assignment. Will keep trying.
+SET send_logs_level='fatal';
+
+DROP TABLE IF EXISTS kafka_test;
+CREATE TABLE kafka_test
+(
+    `raw_message` String
+)
+ENGINE = Kafka
+SETTINGS kafka_broker_list = '0.0.0.0:9092', kafka_group_name = 'test', kafka_topic_list = 'kafka_test', kafka_format = 'RawBLOB', kafka_consumers_pool_ttl_ms=500;
+
+SELECT * FROM kafka_test LIMIT 1 settings stream_like_engine_allow_direct_select=1;
+DROP TABLE kafka_test SYNC;
diff --git a/tests/queries/0_stateless/03523_storage_kafka_shutdown_smoke_local.reference b/tests/queries/0_stateless/03523_storage_kafka_shutdown_smoke_local.reference
diff --git a/tests/queries/0_stateless/03523_storage_kafka_shutdown_smoke_local.sh b/tests/queries/0_stateless/03523_storage_kafka_shutdown_smoke_local.sh
@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+# Tags: no-fasttest
+# Tag no-fasttest -- requires Kafka
+
+# Regression test for proper StorageKafka shutdown
+# https://github.com/ClickHouse/ClickHouse/issues/80674
+#
+# NOTE: this test differs from 03522_storage_kafka_shutdown_smoke, since it creates topic w/o topic group (using named collections)
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+$CLICKHOUSE_LOCAL -nm -q "
+CREATE NAMED COLLECTION kafka_config AS kafka_broker_list = '0.0.0.0:9092';
+
+CREATE TABLE dummy
+(
+    raw_message String
+)
+ENGINE = Kafka(kafka_config)
+SETTINGS kafka_topic_list = 'dummy', kafka_format = 'RawBLOB', kafka_consumers_pool_ttl_ms=500;
+
+SELECT * FROM dummy LIMIT 1 settings stream_like_engine_allow_direct_select=1; -- { serverError 1001 }
+DROP TABLE dummy;
+"
diff --git a/tests/tsan_ignorelist.txt b/tests/tsan_ignorelist.txt
@@ -11,3 +11,13 @@
 [thread]
 # https://github.com/ClickHouse/ClickHouse/issues/55629
 fun:rd_kafka_broker_set_nodename
+# cJSON is used in two libraries (at least) in ClickHouse:
+# - librdkafka
+# - aws-c-common
+#
+# Both libraries has it's own hooks, passed to cJSON_InitHooks, but they are
+# compatible, and eventually they simply call malloc()/free().
+# So let's suppress this warning until it will be fixed properly.
+#
+# See https://github.com/ClickHouse/ClickHouse/issues/80866 for more details.
+fun:cJSON_InitHooks
