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

Close reader stream (#12167) 

Signed-off-by: godchen <qingxiang.chen@zilliz.com>
pull/12163/merge
godchen 2021-11-22 16:17:14 +08:00 committed by GitHub
parent 99b3c56b3e
commit 02bf77aabf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/querynode/segment_loader.go
View File

@ -448,7 +448,7 @@ func (loader *segmentLoader) loadDeltaLogs(segment *Segment, deltaLogs []*datapb
}
func (loader *segmentLoader) FromDmlCPLoadDelete(ctx context.Context, collectionID int64, position *internalpb.MsgPosition) error {
log.Debug("from dmlcp load delete", zap.Any("position", position), zap.Any("msg id", position.MsgID))
log.Debug("from dml check point load delete", zap.Any("position", position), zap.Any("msg id", position.MsgID))
stream, err := loader.factory.NewMsgStream(ctx)
if err != nil {
return err
@ -472,7 +472,6 @@ func (loader *segmentLoader) FromDmlCPLoadDelete(ctx context.Context, collection
if tsMsg == nil {
break
}
log.Debug("receive msg", zap.Any("type", tsMsg.Type()))
if tsMsg.Type() == commonpb.MsgType_Delete {
dmsg := tsMsg.(*msgstream.DeleteMsg)
@ -499,6 +498,8 @@ func (loader *segmentLoader) FromDmlCPLoadDelete(ctx context.Context, collection
go deletePk(loader.historicalReplica, delData, segmentID, &wg)
}
wg.Wait()
stream.Close()
log.Debug("from dml check point load done")
return nil
}