Close the old connection to make sure the broker drops the producer on its side

Author: wenbingshen

pulsar/producer_partition.go

@@ -353,6 +353,15 @@ func (p *partitionProducer) reconnectToBroker() {
 			p.log.WithField("cnx", p.cnx.ID()).Info("Reconnected producer to broker")
 			return
 		}
+
+		if p.cnx != nil {
+            p.cnx.UnregisterListener(p.producerID)
+            state := p.cnx.getState()
+            if state == connectionClosing || state == connectionClosed {
+                p.cnx.close()
+            }
+        }
+
 		errMsg := err.Error()
 		if strings.Contains(errMsg, errTopicNotFount) {
 			// when topic is deleted, we should give up reconnection.