Backport #57822 to 23.10: Disable system.kafka_consumers by default (due to possible live memory leak)

robot-clickhouse · robot-clickhouse · commit 8881fb68d935 · 2023-12-27T17:07:43.000Z
diff --git a/src/Storages/Kafka/StorageKafka.cpp b/src/Storages/Kafka/StorageKafka.cpp
@@ -661,10 +661,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
`@@ -661,10 +661,19 @@ void StorageKafka::updateConfiguration(cppkafka::Configuration & kafka_config,`
`661`	`661`
`662`	`662`	`if (kafka_consumer_weak_ptr_ptr)`
`663`	`663`	`{`
	`664`	`+ /// NOTE: statistics should be consumed, otherwise it creates too much`
	`665`	`+ /// entries in the queue, that leads to memory leak and slow shutdown.`
	`666`	`+ ///`
	`667`	`+ /// This is the case when you have kafka table but no SELECT from it or`
	`668`	`+ /// materialized view attached.`
	`669`	`+ ///`
	`670`	`+ /// So for now it is disabled by default, until properly fixed.`
	`671`	`+#if 0`
`664`	`672`	`if (!config.has(config_prefix + "." + "statistics_interval_ms"))`
`665`	`673`	`{`
`666`	`674`	`kafka_config.set("statistics.interval.ms", "3000"); // every 3 seconds by default. set to 0 to disable.`
`667`	`675`	`}`
	`676`	`+#endif`
`668`	`677`
`669`	`678`	`if (kafka_config.get("statistics.interval.ms") != "0")`
`670`	`679`	`{`