Closed luyui closed 4 years ago
Seems I might have found some clue.
When the consume orderly service tries to consume the queue(ConsumeMessageOrderlyService.cpp:183):
request->takeMessages(msgs, 1);
it may move the only one pending message from m_msgTreeMap
into m_msgTreeMapTemp
, which makes the next pull consider all messages been consumed in the queue(DefaultMQPushConsumerImpl.cpp:100, DefaultMQPushConsumerImpl.cpp:122):
pullRequest->getMessage(msgs);
if ((msgs.size() == 0) && (result.nextBeginOffset > 0)) {
m_callbackOwner->updateConsumeOffset(pullRequest->m_messageQueue, result.nextBeginOffset);
}
thus updating the consume progress wrongly.
This logic will only occur when there's olny one message to be consumed in the queue and consumer orderly is used, which matches my use case nicely.
For now I have changed the PullRequest::getMessage
function to return messages in both m_msgTreeMap
and m_msgTreeMapTemp
, is this change sane? Any potential bug here?
I think u are right but use getCacheMsgCount() to check cache is empty is better, and getCacheMsgCount need to change to below:
int PullRequest::getCacheMsgCount() {
boost::lock_guard<boost::mutex> lock(m_pullRequestLock);
return m_msgTreeMap.size() + m_msgTreeMapTemp.size();
}
@fluyu Could you pull a request to fix this issue?
I think u are right but use getCacheMsgCount() to check cache is empty is better, and getCacheMsgCount need to change to below:
int PullRequest::getCacheMsgCount() { boost::lock_guard<boost::mutex> lock(m_pullRequestLock); return m_msgTreeMap.size() + m_msgTreeMapTemp.size(); }
Thanks for the suggestion, followed and verified, works like a charm.
@fluyu Could you pull a request to fix this issue?
My pleasure and done. #324
BUG REPORT
What did you do (The steps to reproduce)?
What did you expect to see? The message should not been commited to the broker and the consumer group should have 1 message delay
What did you see instead? From the console, found the consum offset been updated to the latest, no delay
Please tell us about your environment:
What is your OS?
What is your client version?
What is your RocketMQ version?
Other information (e.g. detailed explanation, logs, related issues, suggestions on how to fix, etc):
Here is the debug log: 14805_rocketmq-cpp.log, the message consume callback is not returned during the whole log time range.
It seems the incorrect consume offset update happened somewhere between line 5906 and 6200