enhance: [2.5] Close component in topological order when unsub channel (#40796) (#40819)

Cherry-pick from master
pr: #40796

Related to #40795

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

Author: congqixia

internal/querynodev2/services.go

@@ -366,10 +366,11 @@ func (node *QueryNode) UnsubDmChannel(ctx context.Context, req *querypb.UnsubDmC
 	defer node.unsubscribingChannels.Remove(req.GetChannelName())
 	delegator, ok := node.delegators.GetAndRemove(req.GetChannelName())
 	if ok {
+		node.pipelineManager.Remove(req.GetChannelName())
+
 		// close the delegator first to block all coming query/search requests
 		delegator.Close()
 
-		node.pipelineManager.Remove(req.GetChannelName())
 		node.manager.Segment.RemoveBy(ctx, segments.WithChannel(req.GetChannelName()), segments.WithType(segments.SegmentTypeGrowing))
 		_, sealed := node.manager.Segment.RemoveBy(ctx, segments.WithChannel(req.GetChannelName()), segments.WithLevel(datapb.SegmentLevel_L0))
 		// node.tSafeManager.Remove(ctx, req.GetChannelName())

internal/util/pipeline/stream_pipeline.go

@@ -71,7 +71,11 @@ func (p *streamPipeline) work() {
 		case <-p.closeCh:
 			log.Ctx(context.TODO()).Debug("stream pipeline input closed")
 			return
-		case msg := <-p.input:
+		case msg, ok := <-p.input:
+			if !ok {
+				log.Ctx(context.TODO()).Debug("stream pipeline input closed")
+				return
+			}
 			p.lastAccessTime.Store(time.Now())
 			log.Ctx(context.TODO()).RatedDebug(10, "stream pipeline fetch msg", zap.Int("sum", len(msg.Msgs)))
 			p.pipeline.inputChannel <- msg
@@ -169,12 +173,15 @@ func (p *streamPipeline) Start() error {
 
 func (p *streamPipeline) Close() {
 	p.closeOnce.Do(func() {
+		// close datasource first
+		p.dispatcher.Deregister(p.vChannel)
+		// close stream input
 		close(p.closeCh)
 		p.closeWg.Wait()
 		if p.scanner != nil {
 			p.scanner.Close()
 		}
-		p.dispatcher.Deregister(p.vChannel)
+		// close the underline pipeline
 		p.pipeline.Close()
 	})
 }