enhance: Improve balance system with type-safe interface, shared helper, and dead code cleanup (#48372)

related: https://github.com/milvus-io/milvus/issues/48373 - Add missing `return` in `GetSegmentTaskNum` for early exit when no filters - Rename `SetAssignedScore` to `AddAssignedScore` to match `+=` semantics - Remove deprecated `rand.Seed` init (Go 1.20+ auto-seeds) - Add `ScoreAwareAssignPolicy` interface to eliminate type assertions - Extract `BalanceReplicaHelper` to deduplicate streaming service checks across balancers - Remove unused `meta` field from RoundRobin/RowCount/ScoreBased balancer structs --------- Signed-off-by: shaoting-huang <shaoting.huang@zilliz.com>
2026-07-21 10:15:43 +00:00 · 2026-04-10 02:23:40 +08:00
parent b4d61e5ecc
commit 80db722759
24 changed files with 246 additions and 261 deletions
@@ -69,6 +69,17 @@ type AssignPolicy interface {
 	AssignChannel(ctx context.Context, collectionID int64, channels []*meta.DmChannel, nodes []int64, forceAssign bool) []ChannelAssignPlan
 }

+// ScoreAwareAssignPolicy extends AssignPolicy with score-based node conversion
+// and score calculation methods. This interface eliminates the need for type
+// assertions when accessing score-specific functionality.
+type ScoreAwareAssignPolicy interface {
+	AssignPolicy
+	ConvertToNodeItemsBySegment(collectionID int64, nodes []int64) map[int64]*NodeItem
+	ConvertToNodeItemsByChannel(collectionID int64, nodes []int64) map[int64]*NodeItem
+	CalculateSegmentScore(s *meta.Segment) float64
+	CalculateChannelScore(ch *meta.DmChannel, currentCollection int64) float64
+}
+
 // AssignPolicyConfig contains common configuration for assignment policies
 type AssignPolicyConfig struct {
 	// BatchSize limits the number of resources to assign in one batch
@@ -243,9 +243,9 @@ func (p *ScoreBasedAssignPolicy) ConvertToNodeItemsBySegment(collectionID int64,
 	delegatorOverloadFactor := params.Params.QueryCoordCfg.DelegatorMemoryOverloadFactor.GetAsFloat()
 	for _, node := range nodeIDs {
 		if allNodeHasMemInfo {
-			nodeScoreMap[node].SetAssignedScore(nodeMemMap[node] * average)
+			nodeScoreMap[node].AddAssignedScore(nodeMemMap[node] * average)
 		} else {
-			nodeScoreMap[node].SetAssignedScore(average)
+			nodeScoreMap[node].AddAssignedScore(average)
 		}

 		// Add delegator overhead
@@ -437,9 +437,9 @@ func (p *ScoreBasedAssignPolicy) ConvertToNodeItemsByChannel(collectionID int64,

 	for _, node := range nodeIDs {
 		if allNodeHasMemInfo {
-			nodeScoreMap[node].SetAssignedScore(nodeMemMap[node] * average)
+			nodeScoreMap[node].AddAssignedScore(nodeMemMap[node] * average)
 		} else {
-			nodeScoreMap[node].SetAssignedScore(average)
+			nodeScoreMap[node].AddAssignedScore(average)
 		}
 	}

@@ -65,7 +65,7 @@ func (b *NodeItem) GetAssignedScore() float64 {
 	return b.AssignedScore
 }

-func (b *NodeItem) SetAssignedScore(delta float64) {
+func (b *NodeItem) AddAssignedScore(delta float64) {
 	b.AssignedScore += delta
 	b.priority = b.getPriority()
 }
@@ -37,7 +37,7 @@ func TestNewNodeItem(t *testing.T) {
 // TestNodeItem_GetPriority tests priority calculation
 func TestNodeItem_GetPriority(t *testing.T) {
 	item := NewNodeItem(100, 1)
-	item.SetAssignedScore(50)
+	item.AddAssignedScore(50)

 	// Priority should be CurrentScore - AssignedScore = 100 - 50 = 50
 	priority := item.getPriority()
@@ -47,7 +47,7 @@ func TestNodeItem_GetPriority(t *testing.T) {
 // TestNodeItem_GetPriorityWithCurrentScoreDelta tests priority calculation with delta
 func TestNodeItem_GetPriorityWithCurrentScoreDelta(t *testing.T) {
 	item := NewNodeItem(100, 1)
-	item.SetAssignedScore(50)
+	item.AddAssignedScore(50)

 	// Current priority: 100 - 50 = 50
 	// With delta 20: (100 + 20) - 50 = 70
@@ -58,7 +58,7 @@ func TestNodeItem_GetPriorityWithCurrentScoreDelta(t *testing.T) {
 // TestNodeItem_AddCurrentScoreDelta tests adding delta to current score
 func TestNodeItem_AddCurrentScoreDelta(t *testing.T) {
 	item := NewNodeItem(100, 1)
-	item.SetAssignedScore(50)
+	item.AddAssignedScore(50)

 	// Initial priority: 100 - 50 = 50
 	assert.Equal(t, 50, item.getPriority())
@@ -72,12 +72,12 @@ func TestNodeItem_AddCurrentScoreDelta(t *testing.T) {
 	assert.Equal(t, 80, item.getPriority())
 }

-// TestNodeItem_SetAssignedScore tests setting assigned score
-func TestNodeItem_SetAssignedScore(t *testing.T) {
+// TestNodeItem_AddAssignedScore tests adding to assigned score
+func TestNodeItem_AddAssignedScore(t *testing.T) {
 	item := NewNodeItem(100, 1)

 	// Set assigned score
-	item.SetAssignedScore(40)
+	item.AddAssignedScore(40)

 	assert.Equal(t, float64(40), item.GetAssignedScore())
 	// Priority: 100 - 40 = 60
@@ -97,7 +97,7 @@ func TestNodeItem_SetPriority(t *testing.T) {
 // TestNodeItem_String tests string representation
 func TestNodeItem_String(t *testing.T) {
 	item := NewNodeItem(100, 1)
-	item.SetAssignedScore(50)
+	item.AddAssignedScore(50)

 	str := item.String()
 	assert.Contains(t, str, "NodeID: 1")
@@ -109,7 +109,7 @@ func TestNodeItem_String(t *testing.T) {
 // TestNodeItem_NegativeScores tests behavior with negative score changes
 func TestNodeItem_NegativeScores(t *testing.T) {
 	item := NewNodeItem(100, 1)
-	item.SetAssignedScore(50)
+	item.AddAssignedScore(50)

 	// Subtract from current score
 	item.AddCurrentScoreDelta(-30)
@@ -172,13 +172,13 @@ func TestPriorityQueue_WithAssignedScore(t *testing.T) {
 	pq := NewPriorityQueue()

 	item1 := NewNodeItem(100, 1)
-	item1.SetAssignedScore(50) // Priority: 100 - 50 = 50
+	item1.AddAssignedScore(50) // Priority: 100 - 50 = 50

 	item2 := NewNodeItem(80, 2)
-	item2.SetAssignedScore(50) // Priority: 80 - 50 = 30
+	item2.AddAssignedScore(50) // Priority: 80 - 50 = 30

 	item3 := NewNodeItem(120, 3)
-	item3.SetAssignedScore(50) // Priority: 120 - 50 = 70
+	item3.AddAssignedScore(50) // Priority: 120 - 50 = 70

 	pq.Push(&item1)
 	pq.Push(&item2)
@@ -30,8 +30,6 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/internal/querycoordv2/task"
-	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
-	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )
@@ -55,10 +53,9 @@ type Balance interface {
 // This balancer is useful for scenarios where uniform distribution is more important than
 // weighted distribution based on actual load.
 type RoundRobinBalancer struct {
+	BalanceReplicaHelper
 	scheduler    task.Scheduler
-	nodeManager  *session.NodeManager
 	dist         *meta.DistributionManager
-	meta         *meta.Meta
 	targetMgr    meta.TargetManagerInterface
 	assignPolicy assign.AssignPolicy
 }
@@ -101,13 +98,7 @@ func (b *RoundRobinBalancer) BalanceReplica(ctx context.Context, replica *meta.R
 // balanceChannels generates channel balance plans for a replica.
 // It requires at least 2 RW nodes to perform balancing.
 func (b *RoundRobinBalancer) balanceChannels(ctx context.Context, replica *meta.Replica) []assign.ChannelAssignPlan {
-	var rwNodes []int64
-	if streamingutil.IsStreamingServiceEnabled() {
-		rwNodes, _ = utils.GetChannelRWAndRONodesFor260(replica, b.nodeManager)
-	} else {
-		rwNodes = replica.GetRWNodes()
-	}
-
+	rwNodes := b.GetRWNodesForChannels(replica)
 	if len(rwNodes) < 2 {
 		return nil
 	}
@@ -226,16 +217,14 @@ func NewRoundRobinBalancer(
 	scheduler task.Scheduler,
 	nodeManager *session.NodeManager,
 	dist *meta.DistributionManager,
-	meta *meta.Meta,
 	targetMgr meta.TargetManagerInterface,
 ) *RoundRobinBalancer {
 	policy := assign.GetGlobalAssignPolicyFactory().GetPolicy(assign.PolicyTypeRoundRobin)
 	return &RoundRobinBalancer{
-		scheduler:    scheduler,
-		nodeManager:  nodeManager,
-		dist:         dist,
-		meta:         meta,
-		targetMgr:    targetMgr,
-		assignPolicy: policy,
+		BalanceReplicaHelper: BalanceReplicaHelper{nodeManager: nodeManager},
+		scheduler:            scheduler,
+		dist:                 dist,
+		targetMgr:            targetMgr,
+		assignPolicy:         policy,
 	}
 }
@@ -59,7 +59,7 @@ func (suite *BalanceTestSuite) SetupTest() {
 	// Initialize global assign policy factory before creating balancer
 	assign.InitGlobalAssignPolicyFactory(suite.mockScheduler, suite.nodeManager, suite.dist, nil, suite.targetMgr)

-	suite.roundRobinBalancer = NewRoundRobinBalancer(suite.mockScheduler, suite.nodeManager, suite.dist, nil, suite.targetMgr)
+	suite.roundRobinBalancer = NewRoundRobinBalancer(suite.mockScheduler, suite.nodeManager, suite.dist, suite.targetMgr)

 	suite.mockScheduler.EXPECT().GetSegmentTaskDelta(mock.Anything, mock.Anything).Return(0).Maybe()
 	suite.mockScheduler.EXPECT().GetChannelTaskDelta(mock.Anything, mock.Anything).Return(0).Maybe()
@@ -41,7 +41,6 @@ type BalancerFactory struct {
 	scheduler   task.Scheduler
 	nodeManager *session.NodeManager
 	dist        *meta.DistributionManager
-	meta        *meta.Meta
 	targetMgr   meta.TargetManagerInterface
 }

@@ -57,11 +56,10 @@ func InitGlobalBalancerFactory(
 	scheduler task.Scheduler,
 	nodeManager *session.NodeManager,
 	dist *meta.DistributionManager,
-	meta *meta.Meta,
 	targetMgr meta.TargetManagerInterface,
 ) {
 	factoryOnce.Do(func() {
-		globalFactory = NewBalancerFactory(scheduler, nodeManager, dist, meta, targetMgr)
+		globalFactory = NewBalancerFactory(scheduler, nodeManager, dist, targetMgr)
 		log.Info("Global balancer factory initialized")
 	})
 }
@@ -84,7 +82,6 @@ func NewBalancerFactory(
 	scheduler task.Scheduler,
 	nodeManager *session.NodeManager,
 	dist *meta.DistributionManager,
-	meta *meta.Meta,
 	targetMgr meta.TargetManagerInterface,
 ) *BalancerFactory {
 	return &BalancerFactory{
@@ -93,7 +90,6 @@ func NewBalancerFactory(
 		scheduler:           scheduler,
 		nodeManager:         nodeManager,
 		dist:                dist,
-		meta:                meta,
 		targetMgr:           targetMgr,
 	}
 }
@@ -115,20 +111,20 @@ func (f *BalancerFactory) GetBalancer() Balance {

 	switch balanceKey {
 	case meta.RoundRobinBalancerName:
-		balancer = NewRoundRobinBalancer(f.scheduler, f.nodeManager, f.dist, f.meta, f.targetMgr)
+		balancer = NewRoundRobinBalancer(f.scheduler, f.nodeManager, f.dist, f.targetMgr)
 	case meta.RowCountBasedBalancerName:
-		balancer = NewRowCountBasedBalancer(f.scheduler, f.nodeManager, f.dist, f.meta, f.targetMgr)
+		balancer = NewRowCountBasedBalancer(f.scheduler, f.nodeManager, f.dist, f.targetMgr)
 	case meta.ScoreBasedBalancerName:
-		balancer = NewScoreBasedBalancer(f.scheduler, f.nodeManager, f.dist, f.meta, f.targetMgr)
+		balancer = NewScoreBasedBalancer(f.scheduler, f.nodeManager, f.dist, f.targetMgr)
 	case meta.MultiTargetBalancerName:
-		balancer = NewMultiTargetBalancer(f.scheduler, f.nodeManager, f.dist, f.meta, f.targetMgr)
+		balancer = NewMultiTargetBalancer(f.scheduler, f.nodeManager, f.dist, f.targetMgr)
 	case meta.ChannelLevelScoreBalancerName:
-		balancer = NewChannelLevelScoreBalancer(f.scheduler, f.nodeManager, f.dist, f.meta, f.targetMgr)
+		balancer = NewChannelLevelScoreBalancer(f.scheduler, f.nodeManager, f.dist, f.targetMgr)
 	default:
 		log.Info("Unknown balancer type, using default",
 			zap.String("requested", balanceKey),
 			zap.String("default", meta.ScoreBasedBalancerName))
-		balancer = NewScoreBasedBalancer(f.scheduler, f.nodeManager, f.dist, f.meta, f.targetMgr)
+		balancer = NewScoreBasedBalancer(f.scheduler, f.nodeManager, f.dist, f.targetMgr)
 	}

 	f.balancerMap[balanceKey] = balancer
@@ -53,11 +53,10 @@ type ChannelLevelScoreBalancer struct {
 func NewChannelLevelScoreBalancer(scheduler task.Scheduler,
 	nodeManager *session.NodeManager,
 	dist *meta.DistributionManager,
-	meta *meta.Meta,
 	targetMgr meta.TargetManagerInterface,
 ) *ChannelLevelScoreBalancer {
 	return &ChannelLevelScoreBalancer{
-		ScoreBasedBalancer: NewScoreBasedBalancer(scheduler, nodeManager, dist, meta, targetMgr),
+		ScoreBasedBalancer: NewScoreBasedBalancer(scheduler, nodeManager, dist, targetMgr),
 		targetMgr:          targetMgr,
 	}
 }
@@ -198,15 +197,7 @@ func (b *ChannelLevelScoreBalancer) genSegmentPlanForOutboundNodes(ctx context.C
 // It identifies segments on nodes with higher-than-average scores and moves them to nodes
 // with lower scores, using the score-based assign policy.
 func (b *ChannelLevelScoreBalancer) genSegmentPlan(ctx context.Context, br *balanceReport, replica *meta.Replica, channelName string, onlineNodes []int64) []assign.SegmentAssignPlan {
-	// Delegate to the assign policy's implementation with safe type assertion
-	policy, ok := b.assignPolicy.(*assign.ScoreBasedAssignPolicy)
-	if !ok {
-		log.Error("invalid policy type for ScoreBasedBalancer",
-			zap.String("expected", "*assign.ScoreBasedAssignPolicy"),
-			zap.String("actual", fmt.Sprintf("%T", b.assignPolicy)))
-		return nil
-	}
-	nodeItemsMap := policy.ConvertToNodeItemsBySegment(replica.GetCollectionID(), onlineNodes)
+	nodeItemsMap := b.assignPolicy.ConvertToNodeItemsBySegment(replica.GetCollectionID(), onlineNodes)
 	for _, item := range nodeItemsMap {
 		br.AddNodeItem(item)
 	}
@@ -244,7 +235,7 @@ func (b *ChannelLevelScoreBalancer) genSegmentPlan(ctx context.Context, br *bala
 		})
 		for _, s := range segments {
 			segmentsToMove = append(segmentsToMove, s)
-			currentScore -= policy.CalculateSegmentScore(s)
+			currentScore -= b.assignPolicy.CalculateSegmentScore(s)
 			if currentScore <= assignedScore {
 				break
 			}
@@ -43,6 +43,7 @@ import (
 type ChannelLevelScoreBalancerTestSuite struct {
 	suite.Suite
 	balancer      *ChannelLevelScoreBalancer
+	meta          *meta.Meta
 	kv            kv.MetaKv
 	broker        *meta.MockBroker
 	mockScheduler *task.MockScheduler
@@ -82,7 +83,8 @@ func (suite *ChannelLevelScoreBalancerTestSuite) SetupTest() {
 	// Initialize global assign policy factory before creating balancer
 	assign.InitGlobalAssignPolicyFactory(suite.mockScheduler, nodeManager, distManager, testMeta, testTarget)

-	suite.balancer = NewChannelLevelScoreBalancer(suite.mockScheduler, nodeManager, distManager, testMeta, testTarget)
+	suite.meta = testMeta
+	suite.balancer = NewChannelLevelScoreBalancer(suite.mockScheduler, nodeManager, distManager, testTarget)

 	suite.mockScheduler.EXPECT().GetSegmentTaskDelta(mock.Anything, mock.Anything).Return(0).Maybe()
 	suite.mockScheduler.EXPECT().GetChannelTaskDelta(mock.Anything, mock.Anything).Return(0).Maybe()
@@ -397,9 +399,9 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestBalanceOneRound() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)

@@ -421,7 +423,7 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestBalanceOneRound() {
 				nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 				nodeInfo.SetState(c.states[i])
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}

 			// 4. balance and verify result
@@ -517,9 +519,9 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestBalanceMultiRound() {
 		collection.LoadPercentage = 100
 		collection.Status = querypb.LoadStatus_Loaded
 		collection.LoadType = querypb.LoadType_LoadCollection
-		balancer.meta.CollectionManager.PutCollection(ctx, collection)
-		balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(balanceCase.collectionIDs[i], balanceCase.collectionIDs[i]))
-		balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(balanceCase.replicaIDs[i], balanceCase.collectionIDs[i],
+		suite.meta.CollectionManager.PutCollection(ctx, collection)
+		suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(balanceCase.collectionIDs[i], balanceCase.collectionIDs[i]))
+		suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(balanceCase.replicaIDs[i], balanceCase.collectionIDs[i],
 			append(balanceCase.nodes, balanceCase.notExistedNodes...)))
 		balancer.targetMgr.UpdateCollectionNextTarget(ctx, balanceCase.collectionIDs[i])
 		balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, balanceCase.collectionIDs[i])
@@ -539,7 +541,7 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestBalanceMultiRound() {
 		})
 		nodeInfo.SetState(balanceCase.states[i])
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, balanceCase.nodes[i])
+		suite.meta.ResourceManager.HandleNodeUp(ctx, balanceCase.nodes[i])
 	}

 	// 4. first round balance
@@ -634,10 +636,10 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestMultiReplicaBalance() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
 			for replicaID, nodes := range c.replicaWithNodes {
-				balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, c.collectionID, nodes))
+				suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, c.collectionID, nodes))
 			}
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)
@@ -661,7 +663,7 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestMultiReplicaBalance() {
 					nodeInfo.UpdateStats(session.WithChannelCnt(len(c.channelDist[nodes[i]])))
 					nodeInfo.SetState(c.states[i])
 					suite.balancer.nodeManager.Add(nodeInfo)
-					suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
+					suite.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
 				}
 			}

@@ -688,7 +690,7 @@ func (suite *ChannelLevelScoreBalancerTestSuite) getCollectionBalancePlans(balan
 	collectionID int64,
 ) ([]assign.SegmentAssignPlan, []assign.ChannelAssignPlan) {
 	ctx := context.Background()
-	replicas := balancer.meta.ReplicaManager.GetByCollection(ctx, collectionID)
+	replicas := suite.meta.ReplicaManager.GetByCollection(ctx, collectionID)
 	segmentPlans, channelPlans := make([]assign.SegmentAssignPlan, 0), make([]assign.ChannelAssignPlan, 0)
 	for _, replica := range replicas {
 		sPlans, cPlans := balancer.BalanceReplica(ctx, replica)
@@ -730,9 +732,9 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestExclusiveChannelBalance_Cha
 	collection := utils.CreateTestCollection(collectionID, int32(1))
 	collection.LoadPercentage = 100
 	collection.Status = querypb.LoadStatus_Loaded
-	balancer.meta.CollectionManager.PutCollection(ctx, collection)
-	balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, partitionID))
-	balancer.meta.ReplicaManager.Spawn(ctx, 1, map[string]int{meta.DefaultResourceGroupName: 1}, []string{"channel1", "channel2"},
+	suite.meta.CollectionManager.PutCollection(ctx, collection)
+	suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, partitionID))
+	suite.meta.ReplicaManager.Spawn(ctx, 1, map[string]int{meta.DefaultResourceGroupName: 1}, []string{"channel1", "channel2"},
 		commonpb.LoadPriority_LOW)
 	balancer.targetMgr.UpdateCollectionNextTarget(ctx, collectionID)
 	balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, collectionID)
@@ -749,11 +751,11 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestExclusiveChannelBalance_Cha
 		// nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 		nodeInfo.SetState(session.NodeStateNormal)
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodeInfo.ID())
+		suite.meta.ResourceManager.HandleNodeUp(ctx, nodeInfo.ID())
 	}
-	utils.RecoverAllCollection(balancer.meta)
+	utils.RecoverAllCollection(suite.meta)

-	replica := balancer.meta.ReplicaManager.GetByCollection(ctx, collectionID)[0]
+	replica := suite.meta.ReplicaManager.GetByCollection(ctx, collectionID)[0]
 	ch1Nodes := replica.GetChannelRWNodes("channel1")
 	ch2Nodes := replica.GetChannelRWNodes("channel2")
 	suite.Len(ch1Nodes, 2)
@@ -807,9 +809,9 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestExclusiveChannelBalance_Seg
 	collection := utils.CreateTestCollection(collectionID, int32(1))
 	collection.LoadPercentage = 100
 	collection.Status = querypb.LoadStatus_Loaded
-	balancer.meta.CollectionManager.PutCollection(ctx, collection)
-	balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, partitionID))
-	balancer.meta.ReplicaManager.Spawn(ctx, 1, map[string]int{meta.DefaultResourceGroupName: 1}, []string{"channel1", "channel2"}, commonpb.LoadPriority_LOW)
+	suite.meta.CollectionManager.PutCollection(ctx, collection)
+	suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, partitionID))
+	suite.meta.ReplicaManager.Spawn(ctx, 1, map[string]int{meta.DefaultResourceGroupName: 1}, []string{"channel1", "channel2"}, commonpb.LoadPriority_LOW)
 	balancer.targetMgr.UpdateCollectionNextTarget(ctx, collectionID)
 	balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, collectionID)

@@ -825,11 +827,11 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestExclusiveChannelBalance_Seg
 		// nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 		nodeInfo.SetState(session.NodeStateNormal)
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodeInfo.ID())
+		suite.meta.ResourceManager.HandleNodeUp(ctx, nodeInfo.ID())
 	}
-	utils.RecoverAllCollection(balancer.meta)
+	utils.RecoverAllCollection(suite.meta)

-	replica := balancer.meta.ReplicaManager.GetByCollection(ctx, collectionID)[0]
+	replica := suite.meta.ReplicaManager.GetByCollection(ctx, collectionID)[0]
 	ch1Nodes := replica.GetChannelRWNodes("channel1")
 	ch2Nodes := replica.GetChannelRWNodes("channel2")
 	suite.Len(ch1Nodes, 2)
@@ -907,9 +909,9 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestExclusiveChannelBalance_Seg
 	collection := utils.CreateTestCollection(collectionID, int32(1))
 	collection.LoadPercentage = 100
 	collection.Status = querypb.LoadStatus_Loaded
-	balancer.meta.CollectionManager.PutCollection(ctx, collection)
-	balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, partitionID))
-	balancer.meta.ReplicaManager.Spawn(ctx, 1, map[string]int{meta.DefaultResourceGroupName: 1}, []string{"channel1", "channel2"}, commonpb.LoadPriority_LOW)
+	suite.meta.CollectionManager.PutCollection(ctx, collection)
+	suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, partitionID))
+	suite.meta.ReplicaManager.Spawn(ctx, 1, map[string]int{meta.DefaultResourceGroupName: 1}, []string{"channel1", "channel2"}, commonpb.LoadPriority_LOW)
 	balancer.targetMgr.UpdateCollectionNextTarget(ctx, collectionID)
 	balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, collectionID)

@@ -925,11 +927,11 @@ func (suite *ChannelLevelScoreBalancerTestSuite) TestExclusiveChannelBalance_Seg
 		// nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 		nodeInfo.SetState(session.NodeStateNormal)
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodeInfo.ID())
+		suite.meta.ResourceManager.HandleNodeUp(ctx, nodeInfo.ID())
 	}
-	utils.RecoverAllCollection(balancer.meta)
+	utils.RecoverAllCollection(suite.meta)

-	replica := balancer.meta.ReplicaManager.GetByCollection(ctx, collectionID)[0]
+	replica := suite.meta.ReplicaManager.GetByCollection(ctx, collectionID)[0]
 	ch1Nodes := replica.GetChannelRWNodes("channel1")
 	ch2Nodes := replica.GetChannelRWNodes("channel2")
 	suite.Len(ch1Nodes, 2)
@@ -0,0 +1,51 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package balance
+
+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"
+)
+
+// BalanceReplicaHelper provides shared helper methods for getting RW/RO nodes
+// from a replica, handling the streaming service compatibility logic in one place.
+type BalanceReplicaHelper struct {
+	nodeManager *session.NodeManager
+}
+
+// GetRWNodesForChannels returns the RW nodes for channel balancing.
+// When streaming service is enabled, it uses the compatibility helper;
+// otherwise it returns the replica's RW nodes directly.
+func (h *BalanceReplicaHelper) GetRWNodesForChannels(replica *meta.Replica) []int64 {
+	if streamingutil.IsStreamingServiceEnabled() {
+		rwNodes, _ := utils.GetChannelRWAndRONodesFor260(replica, h.nodeManager)
+		return rwNodes
+	}
+	return replica.GetRWNodes()
+}
+
+// GetRWAndRONodesForChannels returns both RW and RO nodes for channel balancing.
+// When streaming service is enabled, it uses the compatibility helper;
+// otherwise it returns the replica's RW and RO nodes directly.
+func (h *BalanceReplicaHelper) GetRWAndRONodesForChannels(replica *meta.Replica) (rwNodes []int64, roNodes []int64) {
+	if streamingutil.IsStreamingServiceEnabled() {
+		return utils.GetChannelRWAndRONodesFor260(replica, h.nodeManager)
+	}
+	return replica.GetRWNodes(), replica.GetRONodes()
+}
@@ -8,7 +8,6 @@ import (
 	"math"
 	"math/rand"
 	"sort"
-	"time"

 	"github.com/samber/lo"
 	"go.uber.org/zap"
@@ -18,17 +17,11 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/params"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/internal/querycoordv2/task"
-	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
-	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )

-func init() {
-	rand.Seed(time.Now().UnixNano())
-}
-
 // rowCountCostModel calculates the cost based on row count distribution across nodes.
 // A lower cost indicates a more balanced distribution of rows.
 type rowCountCostModel struct {
@@ -557,13 +550,7 @@ func (b *MultiTargetBalancer) BalanceReplica(ctx context.Context, replica *meta.
 // balanceChannels generates channel balance plans for a replica.
 // It requires at least 2 RW nodes to perform balancing.
 func (b *MultiTargetBalancer) balanceChannels(ctx context.Context, br *balanceReport, replica *meta.Replica) []assign.ChannelAssignPlan {
-	var rwNodes []int64
-	if streamingutil.IsStreamingServiceEnabled() {
-		rwNodes, _ = utils.GetChannelRWAndRONodesFor260(replica, b.nodeManager)
-	} else {
-		rwNodes = replica.GetRWNodes()
-	}
-
+	rwNodes := b.GetRWNodesForChannels(replica)
 	if len(rwNodes) < 2 {
 		br.AddRecord(StrRecord("no enough rwNodes to balance channels"))
 		return nil
@@ -642,9 +629,9 @@ func (b *MultiTargetBalancer) genPlanByDistributions(nodeSegments, globalNodeSeg

 // NewMultiTargetBalancer creates a new MultiTargetBalancer instance.
 // It embeds a ScoreBasedBalancer and adds multi-objective optimization capabilities.
-func NewMultiTargetBalancer(scheduler task.Scheduler, nodeManager *session.NodeManager, dist *meta.DistributionManager, meta *meta.Meta, targetMgr meta.TargetManagerInterface) *MultiTargetBalancer {
+func NewMultiTargetBalancer(scheduler task.Scheduler, nodeManager *session.NodeManager, dist *meta.DistributionManager, targetMgr meta.TargetManagerInterface) *MultiTargetBalancer {
 	return &MultiTargetBalancer{
-		ScoreBasedBalancer: NewScoreBasedBalancer(scheduler, nodeManager, dist, meta, targetMgr),
+		ScoreBasedBalancer: NewScoreBasedBalancer(scheduler, nodeManager, dist, targetMgr),
 		dist:               dist,
 		targetMgr:          targetMgr,
 	}
@@ -29,8 +29,6 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/internal/querycoordv2/task"
-	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
-	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )
@@ -40,10 +38,9 @@ import (
 // attempting to equalize the row count distribution. This is more accurate than
 // round-robin balancing as it accounts for actual data volume rather than just segment count.
 type RowCountBasedBalancer struct {
+	BalanceReplicaHelper
 	scheduler    task.Scheduler
-	nodeManager  *session.NodeManager
 	dist         *meta.DistributionManager
-	meta         *meta.Meta
 	targetMgr    meta.TargetManagerInterface
 	assignPolicy assign.AssignPolicy
 }
@@ -84,13 +81,7 @@ func (b *RowCountBasedBalancer) BalanceReplica(ctx context.Context, replica *met
 // balanceChannels generates channel balance plans for a replica.
 // It requires at least 2 RW nodes to perform balancing.
 func (b *RowCountBasedBalancer) balanceChannels(ctx context.Context, br *balanceReport, replica *meta.Replica) []assign.ChannelAssignPlan {
-	var rwNodes []int64
-	if streamingutil.IsStreamingServiceEnabled() {
-		rwNodes, _ = utils.GetChannelRWAndRONodesFor260(replica, b.nodeManager)
-	} else {
-		rwNodes = replica.GetRWNodes()
-	}
-
+	rwNodes := b.GetRWNodesForChannels(replica)
 	if len(rwNodes) < 2 {
 		br.AddRecord(StrRecord("no enough rwNodes to balance channels"))
 		return nil
@@ -227,16 +218,14 @@ func NewRowCountBasedBalancer(
 	scheduler task.Scheduler,
 	nodeManager *session.NodeManager,
 	dist *meta.DistributionManager,
-	meta *meta.Meta,
 	targetMgr meta.TargetManagerInterface,
 ) *RowCountBasedBalancer {
 	policy := assign.GetGlobalAssignPolicyFactory().GetPolicy(assign.PolicyTypeRowCount)
 	return &RowCountBasedBalancer{
-		scheduler:    scheduler,
-		nodeManager:  nodeManager,
-		dist:         dist,
-		meta:         meta,
-		targetMgr:    targetMgr,
-		assignPolicy: policy,
+		BalanceReplicaHelper: BalanceReplicaHelper{nodeManager: nodeManager},
+		scheduler:            scheduler,
+		dist:                 dist,
+		targetMgr:            targetMgr,
+		assignPolicy:         policy,
 	}
 }
@@ -45,6 +45,7 @@ import (
 type RowCountBasedBalancerTestSuite struct {
 	suite.Suite
 	balancer      *RowCountBasedBalancer
+	meta          *meta.Meta
 	kv            kv.MetaKv
 	broker        *meta.MockBroker
 	mockScheduler *task.MockScheduler
@@ -84,7 +85,8 @@ func (suite *RowCountBasedBalancerTestSuite) SetupTest() {
 	// Initialize global assign policy factory before creating balancer
 	assign.InitGlobalAssignPolicyFactory(suite.mockScheduler, nodeManager, distManager, testMeta, testTarget)

-	suite.balancer = NewRowCountBasedBalancer(suite.mockScheduler, nodeManager, distManager, testMeta, testTarget)
+	suite.meta = testMeta
+	suite.balancer = NewRowCountBasedBalancer(suite.mockScheduler, nodeManager, distManager, testTarget)

 	suite.broker.EXPECT().GetPartitions(mock.Anything, int64(1)).Return([]int64{1}, nil).Maybe()

@@ -279,9 +281,9 @@ func (suite *RowCountBasedBalancerTestSuite) TestBalance() {
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
 			collection.LoadType = querypb.LoadType_LoadCollection
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(1, 1))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(1, 1, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(1, 1))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(1, 1, c.nodes))
 			suite.broker.ExpectedCalls = nil
 			suite.broker.EXPECT().GetRecoveryInfoV2(mock.Anything, int64(1)).Return(nil, segments, nil)
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, int64(1))
@@ -303,9 +305,9 @@ func (suite *RowCountBasedBalancerTestSuite) TestBalance() {
 				nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 				nodeInfo.SetState(c.states[i])
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}
-			utils.RecoverAllCollection(balancer.meta)
+			utils.RecoverAllCollection(suite.meta)

 			segmentPlans, channelPlans := suite.getCollectionBalancePlans(balancer, 1)
 			if !c.multiple {
@@ -317,7 +319,7 @@ func (suite *RowCountBasedBalancerTestSuite) TestBalance() {
 			}

 			for _, node := range c.nodes {
-				balancer.meta.ResourceManager.HandleNodeDown(ctx, node)
+				suite.meta.ResourceManager.HandleNodeDown(ctx, node)
 				balancer.nodeManager.Remove(node)
 				balancer.dist.SegmentDistManager.Update(node)
 				balancer.dist.ChannelDistManager.Update(node)
@@ -357,8 +359,8 @@ func (suite *RowCountBasedBalancerTestSuite) TestBalanceOnLoadingCollection() {
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loading
 			collection.LoadType = querypb.LoadType_LoadCollection
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(1, 1, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(1, 1, c.nodes))
 			for node, s := range c.distributions {
 				balancer.dist.SegmentDistManager.Update(node, s...)
 			}
@@ -373,7 +375,7 @@ func (suite *RowCountBasedBalancerTestSuite) getCollectionBalancePlans(balancer
 	collectionID int64,
 ) ([]assign.SegmentAssignPlan, []assign.ChannelAssignPlan) {
 	ctx := context.Background()
-	replicas := balancer.meta.ReplicaManager.GetByCollection(ctx, collectionID)
+	replicas := suite.meta.ReplicaManager.GetByCollection(ctx, collectionID)
 	segmentPlans, channelPlans := make([]assign.SegmentAssignPlan, 0), make([]assign.ChannelAssignPlan, 0)
 	for _, replica := range replicas {
 		sPlans, cPlans := balancer.BalanceReplica(ctx, replica)
@@ -529,9 +531,9 @@ func (suite *RowCountBasedBalancerTestSuite) TestDisableBalanceChannel() {
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
 			collection.LoadType = querypb.LoadType_LoadCollection
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(1, 1))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(1, 1, append(c.nodes, c.notExistedNodes...)))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(1, 1))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(1, 1, append(c.nodes, c.notExistedNodes...)))
 			suite.broker.ExpectedCalls = nil
 			suite.broker.EXPECT().GetRecoveryInfoV2(mock.Anything, int64(1)).Return(nil, segments, nil)
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, int64(1))
@@ -553,7 +555,7 @@ func (suite *RowCountBasedBalancerTestSuite) TestDisableBalanceChannel() {
 				nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 				nodeInfo.SetState(c.states[i])
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}

 			Params.Save(Params.QueryCoordCfg.AutoBalanceChannel.Key, fmt.Sprint(c.enableBalanceChannel))
@@ -656,10 +658,10 @@ func (suite *RowCountBasedBalancerTestSuite) TestMultiReplicaBalance() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
 			for replicaID, nodes := range c.replicaWithNodes {
-				balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, c.collectionID, nodes))
+				suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, c.collectionID, nodes))
 			}
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)
@@ -683,7 +685,7 @@ func (suite *RowCountBasedBalancerTestSuite) TestMultiReplicaBalance() {
 					nodeInfo.UpdateStats(session.WithChannelCnt(len(c.channelDist[nodes[i]])))
 					nodeInfo.SetState(c.states[i])
 					suite.balancer.nodeManager.Add(nodeInfo)
-					suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
+					suite.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
 				}
 			}

@@ -28,8 +28,6 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/internal/querycoordv2/task"
-	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
-	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )
@@ -40,12 +38,11 @@ import (
 // This approach considers both collection-specific and global workload to achieve
 // comprehensive load balancing.
 type ScoreBasedBalancer struct {
+	BalanceReplicaHelper
 	scheduler    task.Scheduler
-	nodeManager  *session.NodeManager
 	dist         *meta.DistributionManager
-	meta         *meta.Meta
 	targetMgr    meta.TargetManagerInterface
-	assignPolicy assign.AssignPolicy
+	assignPolicy assign.ScoreAwareAssignPolicy
 }

 // NewScoreBasedBalancer creates a new ScoreBasedBalancer instance.
@@ -53,17 +50,15 @@ type ScoreBasedBalancer struct {
 func NewScoreBasedBalancer(scheduler task.Scheduler,
 	nodeManager *session.NodeManager,
 	dist *meta.DistributionManager,
-	meta *meta.Meta,
 	targetMgr meta.TargetManagerInterface,
 ) *ScoreBasedBalancer {
-	policy := assign.GetGlobalAssignPolicyFactory().GetPolicy(assign.PolicyTypeScoreBased)
+	policy := assign.GetGlobalAssignPolicyFactory().GetPolicy(assign.PolicyTypeScoreBased).(assign.ScoreAwareAssignPolicy)
 	return &ScoreBasedBalancer{
-		scheduler:    scheduler,
-		nodeManager:  nodeManager,
-		dist:         dist,
-		meta:         meta,
-		targetMgr:    targetMgr,
-		assignPolicy: policy,
+		BalanceReplicaHelper: BalanceReplicaHelper{nodeManager: nodeManager},
+		scheduler:            scheduler,
+		dist:                 dist,
+		targetMgr:            targetMgr,
+		assignPolicy:         policy,
 	}
 }

@@ -109,12 +104,7 @@ func (b *ScoreBasedBalancer) BalanceReplica(ctx context.Context, replica *meta.R
 // balanceChannels generates channel balance plans for a replica.
 // It requires at least 2 RW nodes to perform balancing.
 func (b *ScoreBasedBalancer) balanceChannels(ctx context.Context, br *balanceReport, replica *meta.Replica) []assign.ChannelAssignPlan {
-	var rwNodes []int64
-	if streamingutil.IsStreamingServiceEnabled() {
-		rwNodes, _ = utils.GetChannelRWAndRONodesFor260(replica, b.nodeManager)
-	} else {
-		rwNodes = replica.GetRWNodes()
-	}
+	rwNodes := b.GetRWNodesForChannels(replica)
 	if len(rwNodes) < 2 {
 		br.AddRecord(StrRecord("no enough rwNodes to balance channels"))
 		return nil
@@ -140,15 +130,7 @@ func (b *ScoreBasedBalancer) balanceSegments(ctx context.Context, br *balanceRep
 // to nodes with lower scores. Redundant segments (appearing multiple times in distribution)
 // are skipped to avoid conflicts.
 func (b *ScoreBasedBalancer) genSegmentPlan(ctx context.Context, br *balanceReport, replica *meta.Replica, onlineNodes []int64) []assign.SegmentAssignPlan {
-	// Delegate to the assign policy's implementation with safe type assertion
-	policy, ok := b.assignPolicy.(*assign.ScoreBasedAssignPolicy)
-	if !ok {
-		log.Error("invalid policy type for ScoreBasedBalancer",
-			zap.String("expected", "*assign.ScoreBasedAssignPolicy"),
-			zap.String("actual", fmt.Sprintf("%T", b.assignPolicy)))
-		return nil
-	}
-	nodeItemsMap := policy.ConvertToNodeItemsBySegment(replica.GetCollectionID(), onlineNodes)
+	nodeItemsMap := b.assignPolicy.ConvertToNodeItemsBySegment(replica.GetCollectionID(), onlineNodes)
 	for _, item := range nodeItemsMap {
 		br.AddNodeItem(item)
 	}
@@ -186,7 +168,7 @@ func (b *ScoreBasedBalancer) genSegmentPlan(ctx context.Context, br *balanceRepo
 			return segments[i].GetNumOfRows() < segments[j].GetNumOfRows()
 		})
 		for _, s := range segments {
-			segmentScore := policy.CalculateSegmentScore(s)
+			segmentScore := b.assignPolicy.CalculateSegmentScore(s)
 			br.AddRecord(StrRecordf("pick segment %d with score %f from node %d", s.ID, segmentScore, node))
 			segmentsToMove = append(segmentsToMove, s)
 			currentScore -= segmentScore
@@ -224,15 +206,7 @@ func (b *ScoreBasedBalancer) genSegmentPlan(ctx context.Context, br *balanceRepo
 // It identifies channels on nodes with scores above their assigned quota and moves them
 // to nodes with lower scores. Redundant channels are skipped to avoid conflicts.
 func (b *ScoreBasedBalancer) genChannelPlan(ctx context.Context, br *balanceReport, replica *meta.Replica, onlineNodes []int64) []assign.ChannelAssignPlan {
-	// Delegate to the assign policy's implementation with safe type assertion
-	policy, ok := b.assignPolicy.(*assign.ScoreBasedAssignPolicy)
-	if !ok {
-		log.Error("invalid policy type for ScoreBasedBalancer",
-			zap.String("expected", "*assign.ScoreBasedAssignPolicy"),
-			zap.String("actual", fmt.Sprintf("%T", b.assignPolicy)))
-		return nil
-	}
-	nodeItemsMap := policy.ConvertToNodeItemsByChannel(replica.GetCollectionID(), onlineNodes)
+	nodeItemsMap := b.assignPolicy.ConvertToNodeItemsByChannel(replica.GetCollectionID(), onlineNodes)
 	// Add nodes to balance report for logging
 	for _, item := range nodeItemsMap {
 		br.AddNodeItem(item)
@@ -265,7 +239,7 @@ func (b *ScoreBasedBalancer) genChannelPlan(ctx context.Context, br *balanceRepo
 		channels = sortIfChannelAtWALLocated(channels)

 		for _, ch := range channels {
-			channelScore := policy.CalculateChannelScore(ch, replica.GetCollectionID())
+			channelScore := b.assignPolicy.CalculateChannelScore(ch, replica.GetCollectionID())
 			br.AddRecord(StrRecordf("pick channel %s with score %f from node %d", ch.GetChannelName(), channelScore, node))
 			channelsToMove = append(channelsToMove, ch)

@@ -45,6 +45,7 @@ import (
 type ScoreBasedBalancerTestSuite struct {
 	suite.Suite
 	balancer      *ScoreBasedBalancer
+	meta          *meta.Meta
 	kv            kv.MetaKv
 	broker        *meta.MockBroker
 	mockScheduler *task.MockScheduler
@@ -84,7 +85,8 @@ func (suite *ScoreBasedBalancerTestSuite) SetupTest() {
 	// Initialize global assign policy factory before creating balancer
 	assign.InitGlobalAssignPolicyFactory(suite.mockScheduler, nodeManager, distManager, testMeta, testTarget)

-	suite.balancer = NewScoreBasedBalancer(suite.mockScheduler, nodeManager, distManager, testMeta, testTarget)
+	suite.meta = testMeta
+	suite.balancer = NewScoreBasedBalancer(suite.mockScheduler, nodeManager, distManager, testTarget)

 	suite.mockScheduler.EXPECT().GetSegmentTaskDelta(mock.Anything, mock.Anything).Return(0).Maybe()
 	suite.mockScheduler.EXPECT().GetChannelTaskDelta(mock.Anything, mock.Anything).Return(0).Maybe()
@@ -306,7 +308,7 @@ func (suite *ScoreBasedBalancerTestSuite) TestAssignSegmentWithGrowing() {
 	ctx := context.Background()

 	paramtable.Get().Save(paramtable.Get().QueryCoordCfg.DelegatorMemoryOverloadFactor.Key, "0.3")
-	suite.balancer.meta.PutCollection(ctx, &meta.Collection{
+	suite.meta.PutCollection(ctx, &meta.Collection{
 		CollectionLoadInfo: &querypb.CollectionLoadInfo{
 			CollectionID: 1,
 		},
@@ -434,9 +436,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceOneRound() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)

@@ -458,9 +460,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceOneRound() {
 				nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 				nodeInfo.SetState(c.states[i])
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}
-			utils.RecoverAllCollection(balancer.meta)
+			utils.RecoverAllCollection(suite.meta)

 			// 4. balance and verify result
 			segmentPlans, channelPlans := suite.getCollectionBalancePlans(balancer, c.collectionID)
@@ -534,9 +536,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestDelegatorPreserveMemory() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
@@ -559,9 +561,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestDelegatorPreserveMemory() {
 				nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 				nodeInfo.SetState(c.states[i])
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}
-			utils.RecoverAllCollection(balancer.meta)
+			utils.RecoverAllCollection(suite.meta)

 			// disable delegator preserve memory
 			paramtable.Get().Save(paramtable.Get().QueryCoordCfg.DelegatorMemoryOverloadFactor.Key, "0")
@@ -638,9 +640,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceWithExecutingTask() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)

@@ -662,9 +664,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceWithExecutingTask() {
 				nodeInfo.UpdateStats(session.WithChannelCnt(len(c.distributionChannels[c.nodes[i]])))
 				nodeInfo.SetState(c.states[i])
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}
-			utils.RecoverAllCollection(balancer.meta)
+			utils.RecoverAllCollection(suite.meta)

 			// set node delta count
 			suite.mockScheduler.ExpectedCalls = nil
@@ -762,9 +764,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceMultiRound() {
 		collection.LoadPercentage = 100
 		collection.Status = querypb.LoadStatus_Loaded
 		collection.LoadType = querypb.LoadType_LoadCollection
-		balancer.meta.CollectionManager.PutCollection(ctx, collection)
-		balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(balanceCase.collectionIDs[i], balanceCase.collectionIDs[i]))
-		balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(balanceCase.replicaIDs[i], balanceCase.collectionIDs[i],
+		suite.meta.CollectionManager.PutCollection(ctx, collection)
+		suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(balanceCase.collectionIDs[i], balanceCase.collectionIDs[i]))
+		suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(balanceCase.replicaIDs[i], balanceCase.collectionIDs[i],
 			append(balanceCase.nodes, balanceCase.notExistedNodes...)))
 		balancer.targetMgr.UpdateCollectionNextTarget(ctx, balanceCase.collectionIDs[i])
 		balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, balanceCase.collectionIDs[i])
@@ -784,7 +786,7 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceMultiRound() {
 		})
 		nodeInfo.SetState(balanceCase.states[i])
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, balanceCase.nodes[i])
+		suite.meta.ResourceManager.HandleNodeUp(ctx, balanceCase.nodes[i])
 	}

 	// 4. first round balance
@@ -879,10 +881,10 @@ func (suite *ScoreBasedBalancerTestSuite) TestMultiReplicaBalance() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
 			for replicaID, nodes := range c.replicaWithNodes {
-				balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, c.collectionID, nodes))
+				suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, c.collectionID, nodes))
 			}
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)
@@ -906,7 +908,7 @@ func (suite *ScoreBasedBalancerTestSuite) TestMultiReplicaBalance() {
 					nodeInfo.UpdateStats(session.WithChannelCnt(len(c.channelDist[nodes[i]])))
 					nodeInfo.SetState(c.states[i])
 					suite.balancer.nodeManager.Add(nodeInfo)
-					suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
+					suite.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
 				}
 			}

@@ -979,9 +981,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestQNMemoryCapacity() {
 			suite.broker.EXPECT().GetPartitions(mock.Anything, c.collectionID).Return([]int64{c.collectionID}, nil).Maybe()
 			collection.LoadPercentage = 100
 			collection.Status = querypb.LoadStatus_Loaded
-			balancer.meta.CollectionManager.PutCollection(ctx, collection)
-			balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
-			balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
+			suite.meta.CollectionManager.PutCollection(ctx, collection)
+			suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(c.collectionID, c.collectionID))
+			suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(c.replicaID, c.collectionID, c.nodes))
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, c.collectionID)
 			balancer.targetMgr.UpdateCollectionNextTarget(ctx, c.collectionID)
@@ -1006,9 +1008,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestQNMemoryCapacity() {
 				nodeInfo.SetState(c.states[i])
 				nodeInfoMap[c.nodes[i]] = nodeInfo
 				suite.balancer.nodeManager.Add(nodeInfo)
-				suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
+				suite.meta.ResourceManager.HandleNodeUp(ctx, c.nodes[i])
 			}
-			utils.RecoverAllCollection(balancer.meta)
+			utils.RecoverAllCollection(suite.meta)

 			// test qn has same memory capacity
 			nodeInfoMap[1].UpdateStats(session.WithMemCapacity(1024))
@@ -1039,7 +1041,7 @@ func (suite *ScoreBasedBalancerTestSuite) getCollectionBalancePlans(balancer *Sc
 	collectionID int64,
 ) ([]assign.SegmentAssignPlan, []assign.ChannelAssignPlan) {
 	ctx := context.Background()
-	replicas := balancer.meta.ReplicaManager.GetByCollection(ctx, collectionID)
+	replicas := suite.meta.ReplicaManager.GetByCollection(ctx, collectionID)
 	segmentPlans, channelPlans := make([]assign.SegmentAssignPlan, 0), make([]assign.ChannelAssignPlan, 0)
 	for _, replica := range replicas {
 		sPlans, cPlans := balancer.BalanceReplica(ctx, replica)
@@ -1067,9 +1069,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceSegmentAndChannel() {
 	suite.broker.EXPECT().GetPartitions(mock.Anything, collectionID).Return([]int64{collectionID}, nil).Maybe()
 	collection.LoadPercentage = 100
 	collection.Status = querypb.LoadStatus_Loaded
-	balancer.meta.CollectionManager.PutCollection(ctx, collection)
-	balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
-	balancer.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, collectionID, nodes))
+	suite.meta.CollectionManager.PutCollection(ctx, collection)
+	suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
+	suite.meta.ReplicaManager.Put(ctx, utils.CreateTestReplica(replicaID, collectionID, nodes))
 	balancer.targetMgr.UpdateCollectionNextTarget(ctx, collectionID)
 	balancer.targetMgr.UpdateCollectionCurrentTarget(ctx, collectionID)

@@ -1082,9 +1084,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceSegmentAndChannel() {
 		})
 		nodeInfo.SetState(states[i])
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
+		suite.meta.ResourceManager.HandleNodeUp(ctx, nodes[i])
 	}
-	utils.RecoverAllCollection(balancer.meta)
+	utils.RecoverAllCollection(suite.meta)

 	// set unbalance segment distribution
 	balancer.dist.SegmentDistManager.Update(1, []*meta.Segment{
@@ -1149,9 +1151,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnMultiCollections()
 		collection := utils.CreateTestCollection(collectionID, int32(1))
 		collection.LoadPercentage = 100
 		collection.Status = querypb.LoadStatus_Loaded
-		balancer.meta.CollectionManager.PutCollection(ctx, collection)
-		balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
-		balancer.meta.ReplicaManager.Spawn(ctx, collectionID, map[string]int{meta.DefaultResourceGroupName: 1}, nil, commonpb.LoadPriority_LOW)
+		suite.meta.CollectionManager.PutCollection(ctx, collection)
+		suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
+		suite.meta.ReplicaManager.Spawn(ctx, collectionID, map[string]int{meta.DefaultResourceGroupName: 1}, nil, commonpb.LoadPriority_LOW)

 		channels := make([]*datapb.VchannelInfo, channelNum)
 		for i := 0; i < channelNum; i++ {
@@ -1173,8 +1175,8 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnMultiCollections()
 	})
 	nodeInfo.SetState(session.NodeStateNormal)
 	suite.balancer.nodeManager.Add(nodeInfo)
-	suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, 1)
-	utils.RecoverAllCollection(balancer.meta)
+	suite.meta.ResourceManager.HandleNodeUp(ctx, 1)
+	utils.RecoverAllCollection(suite.meta)

 	// mock channel distribution
 	channelDist := make([]*meta.DmChannel, 0)
@@ -1207,8 +1209,8 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnMultiCollections()
 		Version:  common.Version,
 	})
 	suite.balancer.nodeManager.Add(nodeInfo2)
-	suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, 2)
-	utils.RecoverAllCollection(balancer.meta)
+	suite.meta.ResourceManager.HandleNodeUp(ctx, 2)
+	utils.RecoverAllCollection(suite.meta)

 	_, channelPlans := suite.getCollectionBalancePlans(balancer, 1)
 	suite.Len(channelPlans, 1)
@@ -1236,9 +1238,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnDifferentQN() {
 	collection := utils.CreateTestCollection(collectionID, int32(1))
 	collection.LoadPercentage = 100
 	collection.Status = querypb.LoadStatus_Loaded
-	balancer.meta.CollectionManager.PutCollection(ctx, collection)
-	balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
-	balancer.meta.ReplicaManager.Spawn(ctx, collectionID, map[string]int{meta.DefaultResourceGroupName: 1}, nil, commonpb.LoadPriority_LOW)
+	suite.meta.CollectionManager.PutCollection(ctx, collection)
+	suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
+	suite.meta.ReplicaManager.Spawn(ctx, collectionID, map[string]int{meta.DefaultResourceGroupName: 1}, nil, commonpb.LoadPriority_LOW)

 	channels := make([]*datapb.VchannelInfo, channelNum)
 	for i := 0; i < channelNum; i++ {
@@ -1259,8 +1261,8 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnDifferentQN() {
 	})
 	nodeInfo.UpdateStats(session.WithMemCapacity(1024))
 	suite.balancer.nodeManager.Add(nodeInfo)
-	suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, 1)
-	utils.RecoverAllCollection(balancer.meta)
+	suite.meta.ResourceManager.HandleNodeUp(ctx, 1)
+	utils.RecoverAllCollection(suite.meta)

 	// mock channel distribution
 	channelDist := make([]*meta.DmChannel, 0)
@@ -1285,8 +1287,8 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnDifferentQN() {
 		Version:  common.Version,
 	})
 	suite.balancer.nodeManager.Add(nodeInfo2)
-	suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, 2)
-	utils.RecoverAllCollection(balancer.meta)
+	suite.meta.ResourceManager.HandleNodeUp(ctx, 2)
+	utils.RecoverAllCollection(suite.meta)

 	paramtable.Get().Save(paramtable.Get().QueryCoordCfg.BalanceChannelBatchSize.Key, "10")
 	defer paramtable.Get().Reset(paramtable.Get().QueryCoordCfg.BalanceChannelBatchSize.Key)
@@ -1314,9 +1316,9 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnChannelExclusive()
 		collection := utils.CreateTestCollection(collectionID, int32(1))
 		collection.LoadPercentage = 100
 		collection.Status = querypb.LoadStatus_Loaded
-		balancer.meta.CollectionManager.PutCollection(ctx, collection)
-		balancer.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
-		balancer.meta.ReplicaManager.Spawn(ctx, collectionID, map[string]int{meta.DefaultResourceGroupName: 1}, nil, commonpb.LoadPriority_LOW)
+		suite.meta.CollectionManager.PutCollection(ctx, collection)
+		suite.meta.CollectionManager.PutPartition(ctx, utils.CreateTestPartition(collectionID, collectionID))
+		suite.meta.ReplicaManager.Spawn(ctx, collectionID, map[string]int{meta.DefaultResourceGroupName: 1}, nil, commonpb.LoadPriority_LOW)

 		channels := make([]*datapb.VchannelInfo, channelNum)
 		for i := 0; i < channelNum; i++ {
@@ -1340,10 +1342,10 @@ func (suite *ScoreBasedBalancerTestSuite) TestBalanceChannelOnChannelExclusive()
 		})
 		nodeInfo.SetState(session.NodeStateNormal)
 		suite.balancer.nodeManager.Add(nodeInfo)
-		suite.balancer.meta.ResourceManager.HandleNodeUp(ctx, nodeID)
+		suite.meta.ResourceManager.HandleNodeUp(ctx, nodeID)
 	}

-	utils.RecoverAllCollection(balancer.meta)
+	utils.RecoverAllCollection(suite.meta)

 	// mock channels on collection-a to node 1
 	collectionID := int64(1)
@@ -26,8 +26,6 @@ import (
 	"github.com/milvus-io/milvus/internal/querycoordv2/assign"
 	"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/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )
@@ -36,10 +34,10 @@ import (
 // to active nodes (RW nodes). It provides a centralized implementation of stopping balance logic
 // that was previously duplicated across multiple balancer implementations.
 type StoppingBalancer struct {
+	BalanceReplicaHelper
 	dist         *meta.DistributionManager
 	targetMgr    meta.TargetManagerInterface
 	assignPolicy assign.AssignPolicy
-	nodeManager  *session.NodeManager
 }

 // NewStoppingBalancer creates a new StoppingBalancer instance
@@ -50,10 +48,10 @@ func NewStoppingBalancer(
 	nodeManager *session.NodeManager,
 ) *StoppingBalancer {
 	return &StoppingBalancer{
-		dist:         dist,
-		targetMgr:    targetMgr,
-		assignPolicy: assignPolicy,
-		nodeManager:  nodeManager,
+		BalanceReplicaHelper: BalanceReplicaHelper{nodeManager: nodeManager},
+		dist:                 dist,
+		targetMgr:            targetMgr,
+		assignPolicy:         assignPolicy,
 	}
 }

@@ -96,12 +94,7 @@ func (b *StoppingBalancer) BalanceReplica(ctx context.Context, replica *meta.Rep
 }

 func (b *StoppingBalancer) balanceChannels(ctx context.Context, br *balanceReport, replica *meta.Replica) []assign.ChannelAssignPlan {
-	var rwNodes, roNodes []int64
-	if streamingutil.IsStreamingServiceEnabled() {
-		rwNodes, roNodes = utils.GetChannelRWAndRONodesFor260(replica, b.nodeManager)
-	} else {
-		rwNodes, roNodes = replica.GetRWNodes(), replica.GetRONodes()
-	}
+	rwNodes, roNodes := b.GetRWAndRONodesForChannels(replica)

 	// If there are no RW nodes or no RO nodes, no stopping balance is needed
 	if len(rwNodes) == 0 || len(roNodes) == 0 {
@@ -59,7 +59,7 @@ func createTestBalanceChecker() *BalanceChecker {

 	// Initialize global balancer factory for testing
 	balance.ResetGlobalBalancerFactoryForTest()
-	balance.InitGlobalBalancerFactory(scheduler, nodeMgr, dist, metaInstance, targetMgr)
+	balance.InitGlobalBalancerFactory(scheduler, nodeMgr, dist, targetMgr)

 	return NewBalanceChecker(metaInstance, dist, targetMgr, nodeMgr, scheduler)
 }
@@ -460,7 +460,7 @@ func TestBalanceChecker_GenerateBalanceTasksFromReplicas_Success(t *testing.T) {
 	}

 	mockBalancer := mockey.Mock((*balance.BalancerFactory).GetBalancer).To(func(*balance.BalancerFactory) balance.Balance {
-		return balance.NewScoreBasedBalancer(nil, nil, nil, nil, nil)
+		return balance.NewScoreBasedBalancer(nil, nil, nil, nil)
 	}).Build()
 	defer mockBalancer.UnPatch()

@@ -547,7 +547,7 @@ func TestBalanceChecker_GenerateBalanceTasksFromReplicas_StoppingBalanceHighPrio
 	}

 	mockBalancer := mockey.Mock((*balance.BalancerFactory).GetBalancer).To(func(*balance.BalancerFactory) balance.Balance {
-		return balance.NewScoreBasedBalancer(nil, nil, nil, nil, nil)
+		return balance.NewScoreBasedBalancer(nil, nil, nil, nil)
 	}).Build()
 	defer mockBalancer.UnPatch()

@@ -606,7 +606,7 @@ func TestBalanceChecker_GenerateBalanceTasksFromReplicas_NormalBalanceLowPriorit
 	}

 	mockBalancer := mockey.Mock((*balance.BalancerFactory).GetBalancer).To(func(*balance.BalancerFactory) balance.Balance {
-		return balance.NewScoreBasedBalancer(nil, nil, nil, nil, nil)
+		return balance.NewScoreBasedBalancer(nil, nil, nil, nil)
 	}).Build()
 	defer mockBalancer.UnPatch()

@@ -81,7 +81,7 @@ func (suite *ControllerBaseTestSuite) SetupTest() {

 	// Initialize global factories before creating checkers
 	assign.InitGlobalAssignPolicyFactory(suite.scheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetManager)
-	balance.InitGlobalBalancerFactory(suite.scheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetManager)
+	balance.InitGlobalBalancerFactory(suite.scheduler, suite.nodeMgr, suite.dist, suite.targetManager)

 	suite.controller = NewCheckerController(suite.meta, suite.dist, suite.targetManager, suite.nodeMgr, suite.scheduler, suite.broker)
 }
@@ -87,7 +87,7 @@ func (suite *CheckerControllerSuite) SetupTest() {

 	// Initialize global factories before creating checkers
 	assign.InitGlobalAssignPolicyFactory(suite.scheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetManager)
-	balance.InitGlobalBalancerFactory(suite.scheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetManager)
+	balance.InitGlobalBalancerFactory(suite.scheduler, suite.nodeMgr, suite.dist, suite.targetManager)

 	suite.controller = NewCheckerController(suite.meta, suite.dist, suite.targetManager, suite.nodeMgr, suite.scheduler, suite.broker)
 }
@@ -124,7 +124,6 @@ func (suite *OpsServiceSuite) SetupTest() {
 		suite.taskScheduler,
 		suite.nodeMgr,
 		suite.dist,
-		suite.meta,
 		suite.targetMgr,
 	)
 	meta.GlobalFailedLoadCache = meta.NewFailedLoadCache()
@@ -595,7 +594,7 @@ func (suite *OpsServiceSuite) TestTransferSegment() {

 	// test segment not exist in current target, expect no task assign and success
 	assign.InitGlobalAssignPolicyFactory(suite.taskScheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetMgr)
-	balance.InitGlobalBalancerFactory(suite.taskScheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetMgr)
+	balance.InitGlobalBalancerFactory(suite.taskScheduler, suite.nodeMgr, suite.dist, suite.targetMgr)
 	resp, err = suite.server.TransferSegment(ctx, &querypb.TransferSegmentRequest{
 		SourceNodeID: nodes[0],
 		TargetNodeID: nodes[1],
@@ -848,7 +847,7 @@ func (suite *OpsServiceSuite) TestTransferChannel() {

 	// test channel not exist in current target, expect no task assign and success
 	assign.InitGlobalAssignPolicyFactory(suite.taskScheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetMgr)
-	balance.InitGlobalBalancerFactory(suite.taskScheduler, suite.nodeMgr, suite.dist, suite.meta, suite.targetMgr)
+	balance.InitGlobalBalancerFactory(suite.taskScheduler, suite.nodeMgr, suite.dist, suite.targetMgr)
 	resp, err = suite.server.TransferChannel(ctx, &querypb.TransferChannelRequest{
 		SourceNodeID: nodes[0],
 		TargetNodeID: nodes[1],
@@ -337,7 +337,7 @@ func (s *Server) initQueryCoord() error {

 	// Init global balancer factory
 	log.Info("init global balancer factory")
-	balance.InitGlobalBalancerFactory(s.taskScheduler, s.nodeMgr, s.dist, s.meta, s.targetMgr)
+	balance.InitGlobalBalancerFactory(s.taskScheduler, s.nodeMgr, s.dist, s.targetMgr)

 	// Init checker controller
 	log.Info("init checker controller")
@@ -251,7 +251,6 @@ func (suite *ServiceSuite) SetupTest() {
 		suite.taskScheduler,
 		suite.nodeMgr,
 		suite.dist,
-		suite.meta,
 		suite.targetMgr,
 	)
 	meta.GlobalFailedLoadCache = meta.NewFailedLoadCache()
@@ -754,7 +754,7 @@ func (scheduler *taskScheduler) GetChannelTaskNum(filters ...TaskFilter) int {

 func (scheduler *taskScheduler) GetSegmentTaskNum(filters ...TaskFilter) int {
 	if len(filters) == 0 {
-		scheduler.segmentTasks.Len()
+		return scheduler.segmentTasks.Len()
 	}

 	// rewrite this with for loop