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

fix: Use `Remove` sync type for distribution removal (#31215) 

See also #31214

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/31170/head
congqixia 2024-03-13 06:11:04 +08:00 committed by GitHub
parent 96cfae55a5
commit 5b51c20293
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 5 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
internal/querycoordv2/task/executor.go
View File

@ -516,7 +516,7 @@ func (ex *Executor) removeDistribution(task *LeaderTask, step int) error {
req := &querypb.SyncDistributionRequest{
Base: commonpbutil.NewMsgBase(
commonpbutil.WithMsgType(commonpb.MsgType_LoadSegments),
commonpbutil.WithMsgType(commonpb.MsgType_SyncDistribution),
commonpbutil.WithMsgID(task.ID()),
),
CollectionID: task.collectionID,
@ -524,24 +524,23 @@ func (ex *Executor) removeDistribution(task *LeaderTask, step int) error {
ReplicaID: task.ReplicaID(),
Actions: []*querypb.SyncAction{
{
Type: querypb.SyncType_Set,
Type: querypb.SyncType_Remove,
SegmentID: action.SegmentID(),
},
},
}
startTs := time.Now()
log.Info("Sync Distribution...")
log.Info("Remove Distribution...")
status, err := ex.cluster.SyncDistribution(task.Context(), task.leaderID, req)
// status, err := ex.cluster.LoadSegments(task.Context(), leaderID, req)
err = merr.CheckRPCCall(status, err)
if err != nil {
log.Warn("failed to sync distribution", zap.Error(err))
log.Warn("failed to remove distribution", zap.Error(err))
return err
}
elapsed := time.Since(startTs)
log.Info("sync distribution done", zap.Duration("elapsed", elapsed))
log.Info("remove distribution done", zap.Duration("elapsed", elapsed))
return nil
}