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

Merge remote-tracking branch 'apache/5.0.0-beta' into 5.0.0-beta-tmp

Browse Source 
# Conflicts:
#	broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
This commit is contained in:
RongtongJin
2022-03-19 09:41:42 +08:00
parent a44699acd6 8fc952d20c
commit 3cd0ec9a83
34 changed files with 1636 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+10 -8
View File
@@ -497,19 +497,21 @@ public class PopMessageProcessor implements NettyRequestProcessor {
return restNum;
}
getMessageTmpResult = this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup()
, topic, queueId, offset,
requestHeader.getMaxMsgNums() - getMessageResult.getMessageMapedList().size(), messageFilter);
, 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())
|| GetMessageStatus.OFFSET_OVERFLOW_BADLY.equals(getMessageTmpResult.getStatus())
|| GetMessageStatus.OFFSET_FOUND_NULL.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 =