[fix][broker] Unregister non-static metrics collectors registered in Prometheus default registry (apache#24257)

Author: lhotari

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -815,6 +815,10 @@ public CompletableFuture<Void> closeAsync() {
         try {
             log.info("Shutting down Pulsar Broker service");
 
+            // unregister non-static metrics collectors
+            pendingTopicLoadRequests.unregister();
+            pendingLookupRequests.unregister();
+
             // unloads all namespaces gracefully without disrupting mutually
             unloadNamespaceBundlesGracefully();
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ObserverGauge.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.stats.prometheus.metrics;
 
 import io.prometheus.client.Collector;
+import io.prometheus.client.CollectorRegistry;
 import io.prometheus.client.GaugeMetricFamily;
 import io.prometheus.client.SimpleCollector;
 import java.util.ArrayList;
@@ -39,9 +40,11 @@ public Builder supplier(Supplier<Number> supplier) {
             return this;
         }
 
-        public ObserverGauge register() {
+        public ObserverGauge register(CollectorRegistry registry) {
             try {
-                return super.register();
+                ObserverGauge instance = super.register(registry);
+                instance.setRegistry(registry);
+                return instance;
             } catch (Exception e) {
                 // Handle double registration errors in tests
                 return create();
@@ -71,6 +74,7 @@ public static Builder build() {
     }
 
     private final Supplier<Number> supplier;
+    private volatile CollectorRegistry registry;
 
     private ObserverGauge(Builder builder) {
         super(builder);
@@ -99,4 +103,15 @@ public List<MetricFamilySamples> collect() {
     public List<MetricFamilySamples> describe() {
         return Collections.singletonList(new GaugeMetricFamily(fullname, help, labelNames));
     }
+
+    protected void setRegistry(CollectorRegistry registry) {
+        this.registry = registry;
+    }
+
+    public void unregister() {
+        if (registry != null) {
+            registry.unregister(this);
+            registry = null;
+        }
+    }
 }