This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new a05a720 fix request offset not changed in some case (#844)
a05a720 is described below
commit a05a720e107d4ccf4511fe433d91feaedd28b193
Author: cserwen <[email protected]>
AuthorDate: Thu Jul 21 23:46:51 2022 +0800
fix request offset not changed in some case (#844)
Co-authored-by: dengzhiwen1 <[email protected]>
---
consumer/push_consumer.go | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index be2cf61..3221c6c 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -770,6 +770,8 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
goto NEXT
}
+ pc.processPullResult(request.mq, result, sd)
+
switch result.Status {
case primitive.PullFound:
rlog.Debug(fmt.Sprintf("Topic: %s, QueueId: %d found
messages.", request.mq.Topic, request.mq.QueueId), nil)
@@ -779,8 +781,6 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
rt := time.Now().Sub(beginTime) / time.Millisecond
pc.stat.increasePullRT(pc.consumerGroup,
request.mq.Topic, int64(rt))
- pc.processPullResult(request.mq, result, sd)
-
msgFounded := result.GetMessageExts()
firstMsgOffset := int64(math.MaxInt64)
if len(msgFounded) != 0 {
