Rocksmq client should be closed when close a msgstream (#13865)

Signed-off-by: fishpenguin <kun.yu@zilliz.com>

internal/msgstream/mq_msgstream.go

@@ -766,6 +766,7 @@ func (ms *MqTtMsgStream) Close() {
 			reader.Close()
 		}
 	}
+	ms.client.Close()
 }
 
 func (ms *MqTtMsgStream) bufMsgPackToChannel() {

internal/util/mqclient/pulsar_client.go

@@ -123,5 +123,6 @@ func (pc *pulsarClient) BytesToMsgID(id []byte) (MessageID, error) {
 }
 
 func (pc *pulsarClient) Close() {
-	pc.client.Close()
+	// FIXME(yukun): pulsar.client is a singleton, so can't invoke this close when server run
+	// pc.client.Close()
 }

internal/util/rocksmq/client/rocksmq/client_impl.go

@@ -183,10 +183,5 @@ func (c *client) Close() {
 	c.closeOnce.Do(func() {
 		close(c.closeCh)
 		c.wg.Wait()
-		if c.server != nil {
-			c.server.Close()
-		}
-		// Wait all consume goroutines exit
-		c.consumerOptions = nil
 	})
 }