fix: sync partitiion stats blocking balance task(milvus-io#33741) (mi…

…lvus-io#33742) related: milvus-io#33741 Signed-off-by: MrPresent-Han <chun.han@zilliz.com> Co-authored-by: wayblink <anyang.wang@zilliz.com>
wayblink · Jul 2, 2024 · 0aa0f59 · 0aa0f59
1 parent 6b348e4
commit 0aa0f59
Show file tree

Hide file tree

Showing 5 changed files with 7 additions and 9 deletions.
diff --git a/internal/datacoord/handler.go b/internal/datacoord/handler.go
@@ -134,6 +134,7 @@ func (h *ServerHandler) GetQueryVChanPositions(channel RWChannel, partitionIDs .
 			zap.String("channel", channel.GetName()),
 			zap.Int("numOfSegments", len(segments)),
 			zap.Int("indexed segment", len(indexedSegments)),
+			zap.Int64("currentPartitionStatsVersion", currentPartitionStatsVersion),
 		)
 		unIndexedIDs := make(typeutil.UniqueSet)
 

diff --git a/internal/querycoordv2/checkers/balance_checker.go b/internal/querycoordv2/checkers/balance_checker.go
@@ -129,7 +129,7 @@ func (b *BalanceChecker) replicasToBalance() []int64 {
 	hasUnbalancedCollection := false
 	for _, cid := range loadedCollections {
 		if b.normalBalanceCollectionsCurrentRound.Contain(cid) {
-			log.Debug("ScoreBasedBalancer has balanced collection, skip balancing in this round",
+			log.Debug("ScoreBasedBalancer is balancing this collection, skip balancing in this round",
 				zap.Int64("collectionID", cid))
 			continue
 		}

diff --git a/internal/querycoordv2/checkers/leader_checker_test.go b/internal/querycoordv2/checkers/leader_checker_test.go
@@ -75,7 +75,7 @@ func (suite *LeaderCheckerTestSuite) SetupTest() {
 
 	distManager := meta.NewDistributionManager()
 	targetManager := meta.NewTargetManager(suite.broker, suite.meta)
-	suite.checker = NewLeaderChecker(suite.meta, distManager, targetManager, suite.nodeMgr, false)
+	suite.checker = NewLeaderChecker(suite.meta, distManager, targetManager, suite.nodeMgr)
 }
 
 func (suite *LeaderCheckerTestSuite) TearDownTest() {
@@ -476,10 +476,6 @@ func (suite *LeaderCheckerTestSuite) TestIgnoreSyncRemovedSegments() {
 
 func (suite *LeaderCheckerTestSuite) TestUpdatePartitionStats() {
 	testChannel := "test-insert-channel"
-	suite.checker.enableSyncPartitionStats = true
-	defer func() {
-		suite.checker.enableSyncPartitionStats = false
-	}()
 	leaderID := int64(2)
 	observer := suite.checker
 	observer.meta.CollectionManager.PutCollection(utils.CreateTestCollection(1, 1))

diff --git a/internal/querycoordv2/task/action.go b/internal/querycoordv2/task/action.go
@@ -17,6 +17,8 @@
 package task
 
 import (
+	"reflect"
+
 	"github.com/samber/lo"
 	"go.uber.org/atomic"
 
@@ -234,6 +236,8 @@ func (action *LeaderAction) IsFinished(distMgr *meta.DistributionManager) bool {
 		return action.rpcReturned.Load() && dist != nil && dist.NodeID == action.Node()
 	case ActionTypeReduce:
 		return action.rpcReturned.Load() && (dist == nil || dist.NodeID != action.Node())
+	case ActionTypeUpdate:
+		return action.rpcReturned.Load() && (dist != nil && reflect.DeepEqual(action.partStatsVersions, view.PartitionStatsVersions))
 	}
 	return false
 }
diff --git a/tests/integration/balance/balance_test.go b/tests/integration/balance/balance_test.go
@@ -53,15 +53,12 @@ func (s *BalanceTestSuit) SetupSuite() {
 
 	// disable compaction
 	paramtable.Get().Save(paramtable.Get().DataCoordCfg.EnableCompaction.Key, "false")
-	// todo @wayblink repair this test
-	// paramtable.Get().Save(paramtable.Get().QueryNodeCfg.EnableSyncPartitionStats.Key, "false")
 
 	s.Require().NoError(s.SetupEmbedEtcd())
 }
 
 func (s *BalanceTestSuit) TearDownSuite() {
 	defer paramtable.Get().Reset(paramtable.Get().DataCoordCfg.EnableCompaction.Key)
-	// defer paramtable.Get().Reset(paramtable.Get().QueryNodeCfg.EnableSyncPartitionStats.Key)
 
 	s.MiniClusterSuite.TearDownSuite()
 }