consumeMessageCurrently maybe cause data lost if retry queue not Available

[shiftyman]

version v2.1.0

push_comsumer.go, line 1037~1040:

                              if offset >= 0 && !pq.IsDroppd() {
				pc.storage.update(mq, int64(offset), true)
			}
			if len(msgBackFailed) > 0 {
				subMsgs = msgBackFailed
				time.Sleep(5 * time.Second)
				goto RETRY
			}

note: goto retry after offset update，if application crash at this moment，msg lost！

[ShannonDing]

yes, it seems something error here. remove Messages consumer succ from pq is right. like this.

for i := 0; i < len(subMsgs); i++ {
	msg := subMsgs[i]
	if !pc.sendMessageBack(mq.BrokerName, msg, concurrentCtx.DelayLevelWhenNextConsume) {
		msg.ReconsumeTimes += 1
		msgBackFailed = append(msgBackFailed, msg)
	}else {
		msgBackSucc =append(msgBackSucc,msg)
	}
}
	
offset := pq.removeMessage(msgBackSucc...)

if offset >= 0 && !pq.IsDroppd() {
	pc.storage.update(mq, int64(offset), true)
}