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

Author: oskarszoon

util/kafka/kafka_consumer.go

@@ -416,10 +416,6 @@ 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,10 +52,6 @@ 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,10 +119,6 @@ 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
 
@@ -203,10 +199,6 @@ 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

@@ -21,8 +21,16 @@ import (
 	"github.com/bsv-blockchain/teranode/util"
 	inmemorykafka "github.com/bsv-blockchain/teranode/util/kafka/in_memory_kafka"
 	"github.com/bsv-blockchain/teranode/util/retry"
+	"github.com/rcrowley/go-metrics"
 )
 
+// init disables go-metrics globally to prevent memory leak from exponential decay sample heap.
+// This must be set before any Sarama clients are created.
+// See: https://github.com/IBM/sarama/issues/1321
+func init() {
+	metrics.UseNilMetrics = true
+}
+
 // KafkaAsyncProducerI defines the interface for asynchronous Kafka producer operations.
 type KafkaAsyncProducerI interface {
 	// Start begins the async producer operation with the given message channel
@@ -191,10 +199,6 @@ 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}