Merge pull request ClickHouse#57822 from azat/kafka-disable-stat

alexey-milovidov · Enmk · commit ecd52ec70d2b · 2023-12-22T17:31:21.000+01:00
Disable system.kafka_consumers by default (due to possible live memory leak)
diff --git a/src/Storages/Kafka/StorageKafka.cpp b/src/Storages/Kafka/StorageKafka.cpp
@@ -660,10 +660,19 @@ void StorageKafka::updateConfiguration(cppkafka::Configuration & kafka_config,
 
     if (kafka_consumer_weak_ptr_ptr)
     {
+        /// NOTE: statistics should be consumed, otherwise it creates too much
+        /// entries in the queue, that leads to memory leak and slow shutdown.
+        ///
+        /// This is the case when you have kafka table but no SELECT from it or
+        /// materialized view attached.
+        ///
+        /// So for now it is disabled by default, until properly fixed.
+#if 0
         if (!config.has(config_prefix + "." + "statistics_interval_ms"))
         {
             kafka_config.set("statistics.interval.ms", "3000"); // every 3 seconds by default. set to 0 to disable.
         }
+#endif
 
         if (kafka_config.get("statistics.interval.ms") != "0")
         {

Original file line number	Diff line number	Diff line change
`@@ -660,10 +660,19 @@ void StorageKafka::updateConfiguration(cppkafka::Configuration & kafka_config,`
`660`	`660`
`661`	`661`	`if (kafka_consumer_weak_ptr_ptr)`
`662`	`662`	`{`
	`663`	`+ /// NOTE: statistics should be consumed, otherwise it creates too much`
	`664`	`+ /// entries in the queue, that leads to memory leak and slow shutdown.`
	`665`	`+ ///`
	`666`	`+ /// This is the case when you have kafka table but no SELECT from it or`
	`667`	`+ /// materialized view attached.`
	`668`	`+ ///`
	`669`	`+ /// So for now it is disabled by default, until properly fixed.`
	`670`	`+#if 0`
`663`	`671`	`if (!config.has(config_prefix + "." + "statistics_interval_ms"))`
`664`	`672`	`{`
`665`	`673`	`kafka_config.set("statistics.interval.ms", "3000"); // every 3 seconds by default. set to 0 to disable.`
`666`	`674`	`}`
	`675`	`+#endif`
`667`	`676`
`668`	`677`	`if (kafka_config.get("statistics.interval.ms") != "0")`
`669`	`678`	`{`