admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-01-07 19:31:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add deferred clean up for ShardCluster (#17094)

Browse Source 
Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2022-05-18 21:51:56 +08:00 committed by GitHub
parent 1cabaa16a1
commit b95681770b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 11 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
internal/querynode/task.go
View File

@ -186,7 +186,7 @@ func (w *watchDmChannelsTask) PreExecute(ctx context.Context) error {
return nil
}
func (w *watchDmChannelsTask) Execute(ctx context.Context) error {
func (w *watchDmChannelsTask) Execute(ctx context.Context) (err error) {
collectionID := w.req.CollectionID
partitionIDs := w.req.GetPartitionIDs()
@ -201,8 +201,7 @@ func (w *watchDmChannelsTask) Execute(ctx context.Context) error {
}
// get all vChannels
vChannels := make([]Channel, 0)
pChannels := make([]Channel, 0)
var vChannels, pChannels []Channel
VPChannels := make(map[string]string) // map[vChannel]pChannel
for _, info := range w.req.Infos {
v := info.ChannelName
@ -234,6 +233,14 @@ func (w *watchDmChannelsTask) Execute(ctx context.Context) error {
w.node.ShardClusterService.addShardCluster(w.req.GetCollectionID(), w.req.GetReplicaID(), vchannel)
}
defer func() {
if err != nil {
for _, vchannel := range vChannels {
w.node.ShardClusterService.releaseShardCluster(vchannel)
}
}
}()
// load growing segments
unFlushedSegments := make([]*queryPb.SegmentLoadInfo, 0)
unFlushedSegmentIDs := make([]UniqueID, 0)
@ -279,7 +286,7 @@ func (w *watchDmChannelsTask) Execute(ctx context.Context) error {
zap.Int64("collectionID", collectionID),
zap.Int64s("unFlushedSegmentIDs", unFlushedSegmentIDs),
)
err := w.node.loader.loadSegment(req, segmentTypeGrowing)
err = w.node.loader.loadSegment(req, segmentTypeGrowing)
if err != nil {
log.Warn(err.Error())
return err