milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix load the same segments multiple times for manual LoadBalance (#16921) 

Signed-off-by: yah01 <yang.cen@zilliz.com>
pull/17034/head
yah01 2022-05-16 15:41:56 +08:00 committed by GitHub
parent 5f1d7c5267
commit e38c6f6c44
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/querycoord/task.go
View File

@ -2068,9 +2068,7 @@ func (lbt *loadBalanceTask) execute(ctx context.Context) error {
log.Info("loadBalanceTask: add a childTask", zap.Int32("task type", int32(internalTask.msgType())), zap.Any("task", internalTask)) log.Info("loadBalanceTask: add a childTask", zap.Int32("task type", int32(internalTask.msgType())), zap.Any("task", internalTask))
} }
log.Info("loadBalanceTask: assign child task done", zap.Int64s("sourceNodeIDs", lbt.SourceNodeIDs)) log.Info("loadBalanceTask: assign child task done", zap.Int64s("sourceNodeIDs", lbt.SourceNodeIDs))
} } else if lbt.triggerCondition == querypb.TriggerCondition_LoadBalance {
if lbt.triggerCondition == querypb.TriggerCondition_LoadBalance {
if err := lbt.checkForManualLoadBalance(); err != nil { if err := lbt.checkForManualLoadBalance(); err != nil {
lbt.setResultInfo(err) lbt.setResultInfo(err)
return err return err
@ -2084,7 +2082,6 @@ func (lbt *loadBalanceTask) execute(ctx context.Context) error {
balancedSegmentInfos := make(map[UniqueID]*querypb.SegmentInfo) balancedSegmentInfos := make(map[UniqueID]*querypb.SegmentInfo)
balancedSegmentIDs := make([]UniqueID, 0) balancedSegmentIDs := make([]UniqueID, 0)
for _, nodeID := range lbt.SourceNodeIDs { for _, nodeID := range lbt.SourceNodeIDs {
nodeExist := lbt.cluster.hasNode(nodeID) nodeExist := lbt.cluster.hasNode(nodeID)
if !nodeExist { if !nodeExist {
@ -2159,7 +2156,6 @@ func (lbt *loadBalanceTask) execute(ctx context.Context) error {
continue continue
} }
for _, replica := range segmentInfo.ReplicaIds {
segmentBingLog := segmentID2Binlog[segmentID] segmentBingLog := segmentID2Binlog[segmentID]
segmentLoadInfo := lbt.broker.generateSegmentLoadInfo(ctx, collectionID, partitionID, segmentBingLog, true, collectionInfo.Schema) segmentLoadInfo := lbt.broker.generateSegmentLoadInfo(ctx, collectionID, partitionID, segmentBingLog, true, collectionInfo.Schema)
msgBase := proto.Clone(lbt.Base).(*commonpb.MsgBase) msgBase := proto.Clone(lbt.Base).(*commonpb.MsgBase)
@ -2169,11 +2165,10 @@ func (lbt *loadBalanceTask) execute(ctx context.Context) error {
Infos: []*querypb.SegmentLoadInfo{segmentLoadInfo}, Infos: []*querypb.SegmentLoadInfo{segmentLoadInfo},
Schema: collectionInfo.Schema, Schema: collectionInfo.Schema,
CollectionID: collectionID, CollectionID: collectionID,
ReplicaID: replica, ReplicaID: lbt.replicaID,
} }
loadSegmentReqs = append(loadSegmentReqs, loadSegmentReq) loadSegmentReqs = append(loadSegmentReqs, loadSegmentReq)
} }
}
for _, info := range dmChannelInfos { for _, info := range dmChannelInfos {
deltaChannel, err := generateWatchDeltaChannelInfo(info) deltaChannel, err := generateWatchDeltaChannelInfo(info)