fix: unittest

Signed-off-by: chyezh <[email protected]>

internal/datanode/channel/channel_manager_test.go

@@ -265,9 +265,14 @@ func (s *ChannelManagerSuite) TestSubmitSkip() {
 func (s *ChannelManagerSuite) TestSubmitWatchAndRelease() {
 	channel := "by-dev-rootcoord-dml-0"
 
+	stream, err := s.pipelineParams.MsgStreamFactory.NewTtMsgStream(context.Background())
+	s.NoError(err)
+	s.NotNil(stream)
+	stream.AsProducer(context.Background(), []string{channel})
+
 	// watch
 	info := GetWatchInfoByOpID(100, channel, datapb.ChannelWatchState_ToWatch)
-	err := s.manager.Submit(info)
+	err = s.manager.Submit(info)
 	s.NoError(err)
 
 	// wait for result

pkg/mq/mqimpl/rocksmq/client/consumer_impl.go

@@ -135,6 +135,11 @@ func (c *consumer) Close() {
 	err := c.client.server.DestroyConsumerGroup(c.topic, c.consumerName)
 	if err != nil {
 		log.Warn("Consumer close failed", zap.String("topicName", c.topic), zap.String("groupName", c.consumerName), zap.Error(err))
+		// TODO: current rocksmq does't promise the msgmutex will be closed in some unittest,
+		// make the consuming goroutine leak.
+		// Here add a dirty way to close it.
+		close(c.msgMutex)
+		return
 	}
 	<-c.stopCh
 }