opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3503] bugfix: the consumeOffset will be set as 0 when getMessage returns null (#3504)

Browse Source
This commit is contained in:
cserwen
2022-03-18 10:49:02 +08:00
committed by GitHub
parent 176e0d5c22
commit 0b7291b35a
1 changed files with 6 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+6 -4
View File
@@ -498,17 +498,19 @@ public class PopMessageProcessor implements NettyRequestProcessor {
getMessageTmpResult = this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup()
, topic, queueId, offset,
requestHeader.getMaxMsgNums() - getMessageResult.getMessageMapedList().size(), messageFilter);
if (getMessageTmpResult == null) {
return this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;
}
// maybe store offset is not correct.
if (getMessageTmpResult == null
|| GetMessageStatus.OFFSET_TOO_SMALL.equals(getMessageTmpResult.getStatus())
if (GetMessageStatus.OFFSET_TOO_SMALL.equals(getMessageTmpResult.getStatus())
|| GetMessageStatus.OFFSET_OVERFLOW_BADLY.equals(getMessageTmpResult.getStatus())
|| GetMessageStatus.OFFSET_FOUND_NULL.equals(getMessageTmpResult.getStatus())) {
// commit offset, because the offset is not correct
// If offset in store is greater than cq offset, it will cause duplicate messages,
// because offset in PopBuffer is not committed.
POP_LOGGER.warn("Pop initial offset, because store is no correct, {}, {}->{}",
lockKey, offset, getMessageTmpResult != null ? getMessageTmpResult.getNextBeginOffset() : "null");
offset = getMessageTmpResult != null ? getMessageTmpResult.getNextBeginOffset() : 0;
lockKey, offset, getMessageTmpResult.getNextBeginOffset());
offset = getMessageTmpResult.getNextBeginOffset();
this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(), requestHeader.getConsumerGroup(), topic,
queueId, offset);
getMessageTmpResult =