commit ecfd7d3a598a54288817c038cb13f25c26bd370f
author fluyu <github@fluyu.com> Fri Jul 24 19:51:16 2020 +0800
committer James Yin <ywhjames@hotmail.com> Sat Jul 25 05:58:47 2020 +0800
tree aa039847ae987ae16216184f2024476a21082451
parent 999755b68c545c0567670fffd8a0c7b3b28b450a

fix: orderly pusher consumer may commit queue offset too early

src/consumer/DefaultMQPushConsumerImpl.cpp

diff --git a/src/consumer/DefaultMQPushConsumerImpl.cpp b/src/consumer/DefaultMQPushConsumerImpl.cpp
index 1aa0b76..4a79595 100644
--- a/src/consumer/DefaultMQPushConsumerImpl.cpp
+++ b/src/consumer/DefaultMQPushConsumerImpl.cpp
@@ -96,9 +96,7 @@
         }
         pullRequest->setNextOffset(result.nextBeginOffset);
 
-        vector<MQMessageExt> msgs;
-        pullRequest->getMessage(msgs);
-        if ((msgs.size() == 0) && (result.nextBeginOffset > 0)) {
+        if ((pullRequest->getCacheMsgCount() == 0) && (result.nextBeginOffset > 0)) {
           m_callbackOwner->updateConsumeOffset(pullRequest->m_messageQueue, result.nextBeginOffset);
         }
         if (bProducePullRequest) {
@@ -118,9 +116,7 @@
         }
         pullRequest->setNextOffset(result.nextBeginOffset);
 
-        vector<MQMessageExt> msgs;
-        pullRequest->getMessage(msgs);
-        if ((msgs.size() == 0) && (result.nextBeginOffset > 0)) {
+        if ((pullRequest->getCacheMsgCount() == 0) && (result.nextBeginOffset > 0)) {
           m_callbackOwner->updateConsumeOffset(pullRequest->m_messageQueue, result.nextBeginOffset);
         }
         if (bProducePullRequest) {
@@ -740,9 +736,7 @@
           break;
         }
         request->setNextOffset(pullResult.nextBeginOffset);
-        vector<MQMessageExt> msgs;
-        request->getMessage(msgs);
-        if ((msgs.size() == 0) && (pullResult.nextBeginOffset > 0)) {
+        if ((request->getCacheMsgCount() == 0) && (pullResult.nextBeginOffset > 0)) {
           updateConsumeOffset(messageQueue, pullResult.nextBeginOffset);
         }
         producePullMsgTask(request);
@@ -756,9 +750,7 @@
           break;
         }
         request->setNextOffset(pullResult.nextBeginOffset);
-        vector<MQMessageExt> msgs;
-        request->getMessage(msgs);
-        if ((msgs.size() == 0) && (pullResult.nextBeginOffset > 0)) {
+        if ((request->getCacheMsgCount() == 0) && (pullResult.nextBeginOffset > 0)) {
           updateConsumeOffset(messageQueue, pullResult.nextBeginOffset);
         }
         producePullMsgTask(request);

src/consumer/PullRequest.cpp

diff --git a/src/consumer/PullRequest.cpp b/src/consumer/PullRequest.cpp
index 9795880..d86ee7f 100644
--- a/src/consumer/PullRequest.cpp
+++ b/src/consumer/PullRequest.cpp
@@ -92,7 +92,7 @@
 

 int PullRequest::getCacheMsgCount() {

   boost::lock_guard<boost::mutex> lock(m_pullRequestLock);

-  return m_msgTreeMap.size();

+  return m_msgTreeMap.size() + m_msgTreeMapTemp.size();

 }

 

 void PullRequest::getMessageByQueueOffset(vector<MQMessageExt>& msgs, int64 minQueueOffset, int64 maxQueueOffset) {