Close the old connection to make sure the broker drops the producer o…

…n its side

pulsar/internal/connection.go

@@ -843,7 +843,7 @@ func (c *connection) Close() {
 		c.Lock()
 		cnx := c.cnx
 		// do not use changeState() since they share the same lock
-		c.setState(connectionClosed)
+		c.setState(connectionClosing)
 		c.cond.Broadcast()
 		c.Unlock()
 
@@ -853,6 +853,13 @@ func (c *connection) Close() {
 
 		close(c.closeCh)
 
+                c.Lock()
+                cnx := c.cnx
+                // do not use changeState() since they share the same lock
+                c.setState(connectionClosed)
+                c.cond.Broadcast()
+                c.Unlock()
+
 		listeners := make(map[uint64]ConnectionListener)
 		c.listenersLock.Lock()
 		for id, listener := range c.listeners {