Repository: apache/pulsar-client-go
Pull request #826 (2022-08-10 08:40:58)

[client] Add MetricsRegisterer to ClientOptions
Add ClientOptions.MetricsRegisterer for customizing prometheus
metrics registerer.

### Motivation
Currently, client always use `prometheus.DefaultRegisterer` to register metrics, but the application might already have a registerer for metrics registering and exporting.

### Modifications
Add ClientOptions.MetricsRegisterer to allow user to customize the metrics registerer. They can use the existed registerer in their applications.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (no)
  - The public API: (no)
  - The schema: (no)
  - The default values of configurations: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (GoDocs)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/pulsar/client.go b/pulsar/client.go
index bc05b25..9a860ce 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
 	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // NewClient Creates a pulsar client instance
@@ -128,6 +129,10 @@ type ClientOptions struct {
 
 	// Add custom labels to all the metrics reported by this client instance
 	CustomMetricsLabels map[string]string
+
+	// Specify metric registerer used to register metrics.
+	// Default prometheus.DefaultRegisterer
+	MetricsRegisterer prometheus.Registerer
 }
 
 // Client represents a pulsar client
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 09b72af..1bf661e 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -22,6 +22,7 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
@@ -111,11 +112,17 @@ func newClient(options ClientOptions) (Client, error) {
 		options.MetricsCardinality = MetricsCardinalityNamespace
 	}
 
+	if options.MetricsRegisterer == nil {
+		options.MetricsRegisterer = prometheus.DefaultRegisterer
+	}
+
 	var metrics *internal.Metrics
 	if options.CustomMetricsLabels != nil {
-		metrics = internal.NewMetricsProvider(int(options.MetricsCardinality), options.CustomMetricsLabels)
+		metrics = internal.NewMetricsProvider(
+			int(options.MetricsCardinality), options.CustomMetricsLabels, options.MetricsRegisterer)
 	} else {
-		metrics = internal.NewMetricsProvider(int(options.MetricsCardinality), map[string]string{})
+		metrics = internal.NewMetricsProvider(
+			int(options.MetricsCardinality), map[string]string{}, options.MetricsRegisterer)
 	}
 
 	c := &client{
diff --git a/pulsar/internal/metrics.go b/pulsar/internal/metrics.go
index a2af33c..e9bc815 100644
--- a/pulsar/internal/metrics.go
+++ b/pulsar/internal/metrics.go
@@ -96,7 +96,9 @@ type LeveledMetrics struct {
 	ReadersClosed              prometheus.Counter
 }
 
-func NewMetricsProvider(metricsCardinality int, userDefinedLabels map[string]string) *Metrics {
+// NewMetricsProvider returns metrics registered to registerer.
+func NewMetricsProvider(metricsCardinality int, userDefinedLabels map[string]string,
+	registerer prometheus.Registerer) *Metrics {
 	constLabels := map[string]string{
 		"client": "go",
 	}
@@ -329,205 +331,205 @@ func NewMetricsProvider(metricsCardinality int, userDefinedLabels map[string]str
 		}),
 	}
 
-	err := prometheus.DefaultRegisterer.Register(metrics.messagesPublished)
+	err := registerer.Register(metrics.messagesPublished)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.messagesPublished = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.bytesPublished)
+	err = registerer.Register(metrics.bytesPublished)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.bytesPublished = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.messagesPending)
+	err = registerer.Register(metrics.messagesPending)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.messagesPending = are.ExistingCollector.(*prometheus.GaugeVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.bytesPending)
+	err = registerer.Register(metrics.bytesPending)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.bytesPending = are.ExistingCollector.(*prometheus.GaugeVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.publishErrors)
+	err = registerer.Register(metrics.publishErrors)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.publishErrors = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.publishLatency)
+	err = registerer.Register(metrics.publishLatency)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.publishLatency = are.ExistingCollector.(*prometheus.HistogramVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.publishRPCLatency)
+	err = registerer.Register(metrics.publishRPCLatency)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.publishRPCLatency = are.ExistingCollector.(*prometheus.HistogramVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.messagesReceived)
+	err = registerer.Register(metrics.messagesReceived)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.messagesReceived = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.bytesReceived)
+	err = registerer.Register(metrics.bytesReceived)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.bytesReceived = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.prefetchedMessages)
+	err = registerer.Register(metrics.prefetchedMessages)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.prefetchedMessages = are.ExistingCollector.(*prometheus.GaugeVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.prefetchedBytes)
+	err = registerer.Register(metrics.prefetchedBytes)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.prefetchedBytes = are.ExistingCollector.(*prometheus.GaugeVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.acksCounter)
+	err = registerer.Register(metrics.acksCounter)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.acksCounter = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.nacksCounter)
+	err = registerer.Register(metrics.nacksCounter)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.nacksCounter = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.dlqCounter)
+	err = registerer.Register(metrics.dlqCounter)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.dlqCounter = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.processingTime)
+	err = registerer.Register(metrics.processingTime)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.processingTime = are.ExistingCollector.(*prometheus.HistogramVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.producersOpened)
+	err = registerer.Register(metrics.producersOpened)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.producersOpened = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.producersClosed)
+	err = registerer.Register(metrics.producersClosed)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.producersClosed = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.producersReconnectFailure)
+	err = registerer.Register(metrics.producersReconnectFailure)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.producersReconnectFailure = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.producersReconnectMaxRetry)
+	err = registerer.Register(metrics.producersReconnectMaxRetry)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.producersReconnectMaxRetry = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.producersPartitions)
+	err = registerer.Register(metrics.producersPartitions)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.producersPartitions = are.ExistingCollector.(*prometheus.GaugeVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.consumersOpened)
+	err = registerer.Register(metrics.consumersOpened)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.consumersOpened = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.consumersClosed)
+	err = registerer.Register(metrics.consumersClosed)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.consumersClosed = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.consumersReconnectFailure)
+	err = registerer.Register(metrics.consumersReconnectFailure)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.consumersReconnectFailure = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.consumersReconnectMaxRetry)
+	err = registerer.Register(metrics.consumersReconnectMaxRetry)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.consumersReconnectMaxRetry = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.consumersPartitions)
+	err = registerer.Register(metrics.consumersPartitions)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.consumersPartitions = are.ExistingCollector.(*prometheus.GaugeVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.readersOpened)
+	err = registerer.Register(metrics.readersOpened)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.readersOpened = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.readersClosed)
+	err = registerer.Register(metrics.readersClosed)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.readersClosed = are.ExistingCollector.(*prometheus.CounterVec)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.ConnectionsOpened)
+	err = registerer.Register(metrics.ConnectionsOpened)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.ConnectionsOpened = are.ExistingCollector.(prometheus.Counter)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.ConnectionsClosed)
+	err = registerer.Register(metrics.ConnectionsClosed)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.ConnectionsClosed = are.ExistingCollector.(prometheus.Counter)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.ConnectionsEstablishmentErrors)
+	err = registerer.Register(metrics.ConnectionsEstablishmentErrors)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.ConnectionsEstablishmentErrors = are.ExistingCollector.(prometheus.Counter)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.ConnectionsHandshakeErrors)
+	err = registerer.Register(metrics.ConnectionsHandshakeErrors)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.ConnectionsHandshakeErrors = are.ExistingCollector.(prometheus.Counter)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.LookupRequestsCount)
+	err = registerer.Register(metrics.LookupRequestsCount)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.LookupRequestsCount = are.ExistingCollector.(prometheus.Counter)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.PartitionedTopicMetadataRequestsCount)
+	err = registerer.Register(metrics.PartitionedTopicMetadataRequestsCount)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.PartitionedTopicMetadataRequestsCount = are.ExistingCollector.(prometheus.Counter)
 		}
 	}
-	err = prometheus.DefaultRegisterer.Register(metrics.RPCRequestCount)
+	err = registerer.Register(metrics.RPCRequestCount)
 	if err != nil {
 		if are, ok := err.(prometheus.AlreadyRegisteredError); ok {
 			metrics.RPCRequestCount = are.ExistingCollector.(prometheus.Counter)


Test patch:
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index 2255e52..fd50b70 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"github.com/apache/pulsar-client-go/pulsar/internal/crypto"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -33,7 +34,7 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 		eventsCh:             eventsCh,
 		compressionProviders: sync.Map{},
 		options:              &partitionConsumerOpts{},
-		metrics:              internal.NewMetricsProvider(4, map[string]string{}).GetLeveledMetrics("topic"),
+		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
 
@@ -58,6 +59,10 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 	}
 }
 
+func newTestMetrics() *internal.LeveledMetrics {
+	return internal.NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer).GetLeveledMetrics("topic")
+}
+
 func TestBatchMessageIDNoAckTracker(t *testing.T) {
 	eventsCh := make(chan interface{}, 1)
 	pc := partitionConsumer{
@@ -65,7 +70,7 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 		eventsCh:             eventsCh,
 		compressionProviders: sync.Map{},
 		options:              &partitionConsumerOpts{},
-		metrics:              internal.NewMetricsProvider(4, map[string]string{}).GetLeveledMetrics("topic"),
+		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
 
@@ -97,7 +102,7 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 		eventsCh:             eventsCh,
 		compressionProviders: sync.Map{},
 		options:              &partitionConsumerOpts{},
-		metrics:              internal.NewMetricsProvider(4, map[string]string{}).GetLeveledMetrics("topic"),
+		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
 
diff --git a/pulsar/internal/lookup_service_test.go b/pulsar/internal/lookup_service_test.go
index 6bab008..2d07496 100644
--- a/pulsar/internal/lookup_service_test.go
+++ b/pulsar/internal/lookup_service_test.go
@@ -27,6 +27,7 @@ import (
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
@@ -119,7 +120,7 @@ func TestLookupSuccess(t *testing.T) {
 	assert.NoError(t, err)
 	serviceNameResolver := NewPulsarServiceNameResolver(url)
 
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t: t,
 
 		expectedRequests: []pb.CommandLookupTopic{
@@ -138,7 +139,10 @@ func TestLookupSuccess(t *testing.T) {
 				BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
 			},
 		},
-	}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+	}
+
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, serviceNameResolver, false, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -152,8 +156,7 @@ func TestTlsLookupSuccess(t *testing.T) {
 	url, err := url.Parse("pulsar+ssl://example:6651")
 	assert.NoError(t, err)
 	serviceNameResolver := NewPulsarServiceNameResolver(url)
-
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t: t,
 
 		expectedRequests: []pb.CommandLookupTopic{
@@ -172,7 +175,10 @@ func TestTlsLookupSuccess(t *testing.T) {
 				BrokerServiceUrlTls: proto.String("pulsar+ssl://broker-1:6651"),
 			},
 		},
-	}, url, serviceNameResolver, true, "", log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+	}
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+
+	ls := NewLookupService(mockedClient, url, serviceNameResolver, true, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -186,8 +192,7 @@ func TestLookupWithProxy(t *testing.T) {
 	url, err := url.Parse("pulsar://example:6650")
 	assert.NoError(t, err)
 	serviceNameResolver := NewPulsarServiceNameResolver(url)
-
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t: t,
 
 		expectedRequests: []pb.CommandLookupTopic{
@@ -207,7 +212,9 @@ func TestLookupWithProxy(t *testing.T) {
 				ProxyThroughServiceUrl: proto.Bool(true),
 			},
 		},
-	}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+	}
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, serviceNameResolver, false, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -221,7 +228,7 @@ func TestTlsLookupWithProxy(t *testing.T) {
 	url, err := url.Parse("pulsar+ssl://example:6651")
 	assert.NoError(t, err)
 
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t: t,
 
 		expectedRequests: []pb.CommandLookupTopic{
@@ -241,8 +248,10 @@ func TestTlsLookupWithProxy(t *testing.T) {
 				ProxyThroughServiceUrl: proto.Bool(true),
 			},
 		},
-	}, url, NewPulsarServiceNameResolver(url), true, "", log.DefaultNopLogger(),
-		NewMetricsProvider(4, map[string]string{}))
+	}
+	resolver := NewPulsarServiceNameResolver(url)
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, resolver, true, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -256,7 +265,7 @@ func TestLookupWithRedirect(t *testing.T) {
 	url, err := url.Parse("pulsar://example:6650")
 	assert.NoError(t, err)
 
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t:           t,
 		expectedURL: "pulsar://broker-2:6650",
 
@@ -288,8 +297,10 @@ func TestLookupWithRedirect(t *testing.T) {
 				BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
 			},
 		},
-	}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(),
-		NewMetricsProvider(4, map[string]string{}))
+	}
+	resolver := NewPulsarServiceNameResolver(url)
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, resolver, false, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -303,7 +314,7 @@ func TestTlsLookupWithRedirect(t *testing.T) {
 	url, err := url.Parse("pulsar+ssl://example:6651")
 	assert.NoError(t, err)
 
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t:           t,
 		expectedURL: "pulsar+ssl://broker-2:6651",
 
@@ -335,8 +346,11 @@ func TestTlsLookupWithRedirect(t *testing.T) {
 				BrokerServiceUrlTls: proto.String("pulsar+ssl://broker-1:6651"),
 			},
 		},
-	}, url, NewPulsarServiceNameResolver(url), true, "", log.DefaultNopLogger(),
-		NewMetricsProvider(4, map[string]string{}))
+	}
+
+	resolver := NewPulsarServiceNameResolver(url)
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, resolver, true, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -350,7 +364,7 @@ func TestLookupWithInvalidUrlResponse(t *testing.T) {
 	url, err := url.Parse("pulsar://example:6650")
 	assert.NoError(t, err)
 
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t: t,
 
 		expectedRequests: []pb.CommandLookupTopic{
@@ -370,8 +384,10 @@ func TestLookupWithInvalidUrlResponse(t *testing.T) {
 				ProxyThroughServiceUrl: proto.Bool(false),
 			},
 		},
-	}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(),
-		NewMetricsProvider(4, map[string]string{}))
+	}
+	resolver := NewPulsarServiceNameResolver(url)
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, resolver, false, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.Error(t, err)
@@ -382,7 +398,7 @@ func TestLookupWithLookupFailure(t *testing.T) {
 	url, err := url.Parse("pulsar://example:6650")
 	assert.NoError(t, err)
 
-	ls := NewLookupService(&mockedLookupRPCClient{
+	mockedClient := &mockedLookupRPCClient{
 		t: t,
 
 		expectedRequests: []pb.CommandLookupTopic{
@@ -400,8 +416,11 @@ func TestLookupWithLookupFailure(t *testing.T) {
 				Authoritative: proto.Bool(true),
 			},
 		},
-	}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(),
-		NewMetricsProvider(4, map[string]string{}))
+	}
+
+	resolver := NewPulsarServiceNameResolver(url)
+	metricsProvider := NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer)
+	ls := NewLookupService(mockedClient, url, resolver, false, "", log.DefaultNopLogger(), metricsProvider)
 
 	lr, err := ls.Lookup("my-topic")
 	assert.Error(t, err)
@@ -488,7 +507,8 @@ func TestGetPartitionedTopicMetadataSuccess(t *testing.T) {
 				Response:   pb.CommandPartitionedTopicMetadataResponse_Success.Enum(),
 			},
 		},
-	}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+	}, url, serviceNameResolver, false, "", log.DefaultNopLogger(),
+		NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer))
 
 	metadata, err := ls.GetPartitionedTopicMetadata("my-topic")
 	assert.NoError(t, err)
@@ -520,7 +540,8 @@ func TestLookupSuccessWithMultipleHosts(t *testing.T) {
 				BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
 			},
 		},
-	}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+	}, url, serviceNameResolver, false, "", log.DefaultNopLogger(),
+		NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer))
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -581,7 +602,7 @@ func TestHttpLookupSuccess(t *testing.T) {
 	serviceNameResolver := NewPulsarServiceNameResolver(url)
 	httpClient := NewMockHTTPClient(serviceNameResolver)
 	ls := NewHTTPLookupService(httpClient, url, serviceNameResolver, false,
-		log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+		log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer))
 
 	lr, err := ls.Lookup("my-topic")
 	assert.NoError(t, err)
@@ -597,7 +618,7 @@ func TestHttpGetPartitionedTopicMetadataSuccess(t *testing.T) {
 	serviceNameResolver := NewPulsarServiceNameResolver(url)
 	httpClient := NewMockHTTPClient(serviceNameResolver)
 	ls := NewHTTPLookupService(httpClient, url, serviceNameResolver, false,
-		log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}))
+		log.DefaultNopLogger(), NewMetricsProvider(4, map[string]string{}, prometheus.DefaultRegisterer))
 
 	tMetadata, err := ls.GetPartitionedTopicMetadata("my-topic")
 	assert.NoError(t, err)