Disable go-metrics for Kafka to prevent memory leak (#51)

Author: oskarszoon

util/kafka/kafka_consumer.go

@@ -416,6 +416,10 @@ func NewKafkaConsumerGroup(cfg KafkaConsumerConfig) (*KafkaConsumerGroup, error)
 	config := sarama.NewConfig()
 	config.Consumer.Return.Errors = true
 
+	// Disable go-metrics to prevent memory leak from exponential decay sample heap
+	// See: https://github.com/IBM/sarama/issues/1321
+	config.MetricRegistry = nil
+
 	// Enable Sarama debug logging for consumer diagnostics (only if configured)
 	// By default, SARAMA logs are too verbose and not needed in production
 	if cfg.EnableDebugLogging {

util/kafka/kafka_health.go

@@ -52,6 +52,10 @@ func HealthChecker(_ context.Context, brokersURL []string) func(ctx context.Cont
 		config.Net.ReadTimeout = 100 * time.Millisecond
 		config.Net.WriteTimeout = 100 * time.Millisecond
 		config.Metadata.Retry.Max = 0
+
+		// Disable go-metrics to prevent memory leak from exponential decay sample heap
+		// See: https://github.com/IBM/sarama/issues/1321
+		config.MetricRegistry = nil
 		config.Metadata.Full = true
 		config.Metadata.AllowAutoTopicCreation = false
 

util/kafka/kafka_producer.go

@@ -119,6 +119,10 @@ func NewKafkaProducer(kafkaURL *url.URL, kafkaSettings *settings.KafkaSettings)
 	config := sarama.NewConfig()
 	config.Version = sarama.V2_1_0_0
 
+	// Disable go-metrics to prevent memory leak from exponential decay sample heap
+	// See: https://github.com/IBM/sarama/issues/1321
+	config.MetricRegistry = nil
+
 	// Note: Debug logging not supported for sync producer as it doesn't have a logger parameter
 	// If needed, add a logger parameter to NewKafkaProducer function
 
@@ -199,6 +203,10 @@ func ConnectProducer(brokersURL []string, topic string, partitions int32, kafkaS
 	config.Producer.Retry.Max = 5
 	config.Producer.Partitioner = sarama.NewManualPartitioner
 
+	// Disable go-metrics to prevent memory leak from exponential decay sample heap
+	// See: https://github.com/IBM/sarama/issues/1321
+	config.MetricRegistry = nil
+
 	// Apply authentication settings if kafkaSettings provided and TLS is enabled
 	if kafkaSettings != nil && kafkaSettings.EnableTLS {
 		if err := configureKafkaAuthFromFields(config, kafkaSettings.EnableTLS, kafkaSettings.TLSSkipVerify,

util/kafka/kafka_producer_async.go

@@ -191,6 +191,10 @@ func NewKafkaAsyncProducer(logger ulogger.Logger, cfg KafkaProducerConfig) (*Kaf
 	config.Producer.Flush.Frequency = cfg.FlushFrequency
 	// config.Producer.Return.Successes = true
 
+	// Disable go-metrics to prevent memory leak from exponential decay sample heap
+	// See: https://github.com/IBM/sarama/issues/1321
+	config.MetricRegistry = nil
+
 	// Enable Sarama debug logging if configured
 	if cfg.EnableDebugLogging {
 		sarama.Logger = &saramaLoggerAdapter{logger: logger}