Fix producer goroutine leak

Per Producer there is a goroutine leaked. This goroutine is used for the
partition auto-discovery. This will ensure the goroutine has an exit
condition and is cleanup after the `Close()` of a producer.

pulsar/producer_impl.go

@@ -59,6 +59,7 @@ type producer struct {
 	numPartitions uint32
 	messageRouter func(*ProducerMessage, TopicMetadata) int
 	ticker        *time.Ticker
+	tickerStop    chan struct{}
 
 	log *log.Entry
 }
@@ -119,11 +120,17 @@ func newProducer(client *client, options *ProducerOptions) (*producer, error) {
 	}
 
 	p.ticker = time.NewTicker(partitionsAutoDiscoveryInterval)
+	p.tickerStop = make(chan struct{})
 
 	go func() {
-		for range p.ticker.C {
-			p.log.Debug("Auto discovering new partitions")
-			p.internalCreatePartitionsProducers()
+		for {
+			select {
+			case <-p.ticker.C:
+				p.log.Debug("Auto discovering new partitions")
+				p.internalCreatePartitionsProducers()
+			case <-p.tickerStop:
+				return
+			}
 		}
 	}()
 
@@ -282,6 +289,8 @@ func (p *producer) Close() {
 	defer p.RUnlock()
 	if p.ticker != nil {
 		p.ticker.Stop()
+		close(p.tickerStop)
+		p.ticker = nil
 	}
 
 	for _, pp := range p.producers {