test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-01 19:39:21 +08:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

Fix message stream miss ack

Browse Source 
Signed-off-by: xige-16 <xi.ge@zilliz.com>
This commit is contained in:
xige-16 2021-03-02 13:15:49 +08:00 committed by yefu.chen
parent 3e5f05fc0a
commit 795cb74c85
1 changed files with 2 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/msgstream/pulsarms/pulsar_msgstream.go
View File

@ -335,6 +335,7 @@ func (ms *PulsarMsgStream) receiveMsg(consumer Consumer) {
if !ok {
return
}
consumer.Ack(pulsarMsg)
headerMsg := commonpb.MsgHeader{}
err := proto.Unmarshal(pulsarMsg.Payload(), &headerMsg)
if err != nil {
@ -430,6 +431,7 @@ func (ms *PulsarMsgStream) bufMsgPackToChannel() {
msgLen := len(consumer.Chan())
for i := 0; i < msgLen; i++ {
msg := <-consumer.Chan()
consumer.Ack(msg)
pulsarMsgBuffer = append(pulsarMsgBuffer, msg)
}
}