admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-08 01:58:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

add log when seek in msgstream (#6244)

Browse Source 
Signed-off-by: yefu.chen <yefu.chen@zilliz.com>
This commit is contained in:
neza2017 2021-07-01 13:46:16 +08:00 committed by GitHub
parent d7a8c9f552
commit a518e408b1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 4 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/msgstream/mq_msgstream.go
View File

@ -388,15 +388,19 @@ func (ms *mqMsgStream) Seek(msgPositions []*internalpb.MsgPosition) error {
if err != nil { if err != nil {
return err return err
} }
log.Debug("MsgStream begin to seek", zap.Any("MessageID", messageID))
err = consumer.Seek(messageID) err = consumer.Seek(messageID)
if err != nil { if err != nil {
return err return err
} }
log.Debug("MsgStream seek finished", zap.Any("MessageID", messageID))
if _, ok := consumer.(*mqclient.RmqConsumer); !ok { if _, ok := consumer.(*mqclient.RmqConsumer); !ok {
log.Debug("MsgStream begin to read one message after seek")
msg, ok := <-consumer.Chan() msg, ok := <-consumer.Chan()
if !ok { if !ok {
return errors.New("consumer closed") return errors.New("consumer closed")
} }
log.Debug("MsgStream finish reading one message after seek")
consumer.Ack(msg) consumer.Ack(msg)
if !bytes.Equal(msg.ID().Serialize(), messageID.Serialize()) { if !bytes.Equal(msg.ID().Serialize(), messageID.Serialize()) {
err = fmt.Errorf("seek msg not correct") err = fmt.Errorf("seek msg not correct")