Make PartitionsAutoDiscoveryInterval configurable (#514)

Co-authored-by: Chen Liu <cliu@splunk.com>

pulsar/consumer_test.go

@@ -1430,7 +1430,6 @@ func TestConsumerAddTopicPartitions(t *testing.T) {
 	makeHTTPCall(t, http.MethodPut, testURL, "3")
 
 	// create producer
-	partitionsAutoDiscoveryInterval = 100 * time.Millisecond
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic: topic,
 		MessageRouter: func(msg *ProducerMessage, topicMetadata TopicMetadata) int {
@@ -1439,6 +1438,7 @@ func TestConsumerAddTopicPartitions(t *testing.T) {
 			assert.NoError(t, err)
 			return i
 		},
+		PartitionsAutoDiscoveryInterval: 100 * time.Millisecond,
 	})
 	assert.Nil(t, err)
 	defer producer.Close()

pulsar/producer.go

@@ -159,6 +159,10 @@ type ProducerOptions struct {
 	// - DefaultBatchBuilder
 	// - KeyBasedBatchBuilder
 	BatcherBuilderType
+
+	// PartitionsAutoDiscoveryInterval is the time interval for the background process to discover new partitions
+	// Default is 1 minute
+	PartitionsAutoDiscoveryInterval time.Duration
 }
 
 // Producer is used to publish messages on a topic

pulsar/producer_impl.go

@@ -40,6 +40,9 @@ const (
 
 	// defaultMaxMessagesPerBatch init default num of entries in per batch.
 	defaultMaxMessagesPerBatch = 1000
+
+	// defaultPartitionsAutoDiscoveryInterval init default time interval for partitions auto discovery
+	defaultPartitionsAutoDiscoveryInterval = 1 * time.Minute
 )
 
 type producer struct {
@@ -57,8 +60,6 @@ type producer struct {
 	metrics       *internal.TopicMetrics
 }
 
-var partitionsAutoDiscoveryInterval = 1 * time.Minute
-
 func getHashingFunction(s HashingScheme) func(string) uint32 {
 	switch s {
 	case JavaStringHash:
@@ -87,6 +88,9 @@ func newProducer(client *client, options *ProducerOptions) (*producer, error) {
 	if options.BatchingMaxPublishDelay <= 0 {
 		options.BatchingMaxPublishDelay = defaultBatchingMaxPublishDelay
 	}
+	if options.PartitionsAutoDiscoveryInterval <= 0 {
+		options.PartitionsAutoDiscoveryInterval = defaultPartitionsAutoDiscoveryInterval
+	}
 
 	p := &producer{
 		options: options,
@@ -125,7 +129,7 @@ func newProducer(client *client, options *ProducerOptions) (*producer, error) {
 		return nil, err
 	}
 
-	p.stopDiscovery = p.runBackgroundPartitionDiscovery(partitionsAutoDiscoveryInterval)
+	p.stopDiscovery = p.runBackgroundPartitionDiscovery(options.PartitionsAutoDiscoveryInterval)
 
 	p.metrics.ProducersOpened.Inc()
 	return p, nil