fix: add unittest

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

internal/coordinator/snmanager/streaming_node_manager_test.go

@@ -0,0 +1,62 @@
+package snmanager
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_balancer"
+	"github.com/milvus-io/milvus/pkg/streaming/util/types"
+	"github.com/milvus-io/milvus/pkg/util/typeutil"
+)
+
+type pChannelInfoAssigned struct {
+	version   typeutil.VersionInt64Pair
+	pchannels []types.PChannelInfoAssigned
+}
+
+func TestStreamingNodeManager(t *testing.T) {
+	m := newStreamingNodeManager()
+	b := mock_balancer.NewMockBalancer(t)
+
+	ch := make(chan pChannelInfoAssigned, 1)
+	b.EXPECT().WatchChannelAssignments(mock.Anything, mock.Anything).Run(
+		func(ctx context.Context, cb func(typeutil.VersionInt64Pair, []types.PChannelInfoAssigned) error) {
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case p := <-ch:
+					cb(p.version, p.pchannels)
+				}
+			}
+		})
+	m.SetBalancerReady(b)
+
+	streamingNodes := m.GetStreamingQueryNodeIDs()
+	assert.Empty(t, streamingNodes)
+
+	ch <- pChannelInfoAssigned{
+		version: typeutil.VersionInt64Pair{
+			Global: 1,
+			Local:  1,
+		},
+		pchannels: []types.PChannelInfoAssigned{
+			{
+				Channel: types.PChannelInfo{Name: "a_test", Term: 1},
+				Node:    types.StreamingNodeInfo{ServerID: 1, Address: "localhost:1"},
+			},
+		},
+	}
+
+	listener := m.ListenNodeChanged()
+	err := listener.Wait(context.Background())
+	assert.NoError(t, err)
+
+	node := m.GetWALLocated("a_test")
+	assert.Equal(t, node, int64(1))
+	streamingNodes = m.GetStreamingQueryNodeIDs()
+	assert.Equal(t, len(streamingNodes), 1)
+}

internal/querycoordv2/observers/replica_observer.go

@@ -27,6 +27,7 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
 	"github.com/milvus-io/milvus/internal/querycoordv2/params"
 	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
+	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/util/syncutil"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
@@ -57,7 +58,9 @@ func (ob *ReplicaObserver) Start() {
 
 		ob.wg.Add(2)
 		go ob.schedule(ctx)
-		go ob.scheduleStreamingQN(ctx)
+		if streamingutil.IsStreamingServiceEnabled() {
+			go ob.scheduleStreamingQN(ctx)
+		}
 	})
 }
 

internal/querycoordv2/ops_services.go

@@ -28,6 +28,7 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
+	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
@@ -273,7 +274,11 @@ func (s *Server) TransferSegment(ctx context.Context, req *querypb.TransferSegme
 		// when no dst node specified, default to use all other nodes in same
 		dstNodeSet := typeutil.NewUniqueSet()
 		if req.GetToAllNodes() {
-			dstNodeSet.Insert(replica.GetRWNodes()...)
+			if streamingutil.IsStreamingServiceEnabled() {
+				dstNodeSet.Insert(replica.GetRWSQNodes()...)
+			} else {
+				dstNodeSet.Insert(replica.GetRWNodes()...)
+			}
 		} else {
 			// check whether dstNode is healthy
 			if err := s.isStoppingNode(ctx, req.GetTargetNodeID()); err != nil {
@@ -345,7 +350,11 @@ func (s *Server) TransferChannel(ctx context.Context, req *querypb.TransferChann
 		// when no dst node specified, default to use all other nodes in same
 		dstNodeSet := typeutil.NewUniqueSet()
 		if req.GetToAllNodes() {
-			dstNodeSet.Insert(replica.GetRWSQNodes()...)
+			if streamingutil.IsStreamingServiceEnabled() {
+				dstNodeSet.Insert(replica.GetRWSQNodes()...)
+			} else {
+				dstNodeSet.Insert(replica.GetRWNodes()...)
+			}
 		} else {
 			// check whether dstNode is healthy
 			if err := s.isStoppingNode(ctx, req.GetTargetNodeID()); err != nil {

internal/querycoordv2/utils/meta.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/milvus-io/milvus/internal/coordinator/snmanager"
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
+	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
@@ -162,7 +163,9 @@ func SpawnReplicasWithRG(ctx context.Context, m *meta.Meta, collection int64, re
 	}
 	// Active recover it.
 	RecoverReplicaOfCollection(ctx, m, collection)
-	m.RecoverSQNodesInCollection(ctx, collection, snmanager.StaticStreamingNodeManager.GetStreamingQueryNodeIDs())
+	if streamingutil.IsStreamingServiceEnabled() {
+		m.RecoverSQNodesInCollection(ctx, collection, snmanager.StaticStreamingNodeManager.GetStreamingQueryNodeIDs())
+	}
 	return replicas, nil
 }