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

[ISSUE #3949] Set topic for LocalMessageService

Browse Source
This commit is contained in:
zhouxiang
2022-07-19 17:56:28 +08:00
parent 9a0481c0b5
commit f88df48baf
1 changed files with 34 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java
+34 -37
View File
@@ -238,16 +238,14 @@ public class LocalMessageService implements MessageService {
PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) r.readCustomHeader();
if (popStatus == PopStatus.FOUND) {
Map<String, Long> startOffsetInfo = null;
Map<String, List<Long>> msgOffsetInfo = null;
Map<String, Integer> orderCountInfo = null;
if (requestHeader != null) {
popResult.setInvisibleTime(responseHeader.getInvisibleTime());
popResult.setPopTime(responseHeader.getPopTime());
startOffsetInfo = ExtraInfoUtil.parseStartOffsetInfo(responseHeader.getStartOffsetInfo());
msgOffsetInfo = ExtraInfoUtil.parseMsgOffsetInfo(responseHeader.getMsgOffsetInfo());
orderCountInfo = ExtraInfoUtil.parseOrderCountInfo(responseHeader.getOrderCountInfo());
}
Map<String, Long> startOffsetInfo;
Map<String, List<Long>> msgOffsetInfo;
Map<String, Integer> orderCountInfo;
popResult.setInvisibleTime(responseHeader.getInvisibleTime());
popResult.setPopTime(responseHeader.getPopTime());
startOffsetInfo = ExtraInfoUtil.parseStartOffsetInfo(responseHeader.getStartOffsetInfo());
msgOffsetInfo = ExtraInfoUtil.parseMsgOffsetInfo(responseHeader.getMsgOffsetInfo());
orderCountInfo = ExtraInfoUtil.parseOrderCountInfo(responseHeader.getOrderCountInfo());
// <topicMark@queueId, msg queueOffset>
Map<String, List<Long>> sortMap = new HashMap<>(16);
for (MessageExt messageExt : messageExtList) {
@@ -259,38 +257,37 @@ public class LocalMessageService implements MessageService {
}
Map<String, String> map = new HashMap<>(5);
for (MessageExt messageExt : messageExtList) {
if (requestHeader != null) {
if (startOffsetInfo == null) {
// we should set the check point info to extraInfo field , if the command is popMsg
// find pop ck offset
String key = messageExt.getTopic() + messageExt.getQueueId();
if (!map.containsKey(messageExt.getTopic() + messageExt.getQueueId())) {
map.put(key, ExtraInfoUtil.buildExtraInfo(messageExt.getQueueOffset(), responseHeader.getPopTime(), responseHeader.getInvisibleTime(), responseHeader.getReviveQid(),
messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId()));
}
messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
} else {
String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
int index = sortMap.get(key).indexOf(messageExt.getQueueOffset());
Long msgQueueOffset = msgOffsetInfo.get(key).get(index);
if (msgQueueOffset != messageExt.getQueueOffset()) {
log.warn("Queue offset [{}] of msg is strange, not equal to the stored in msg, {}", msgQueueOffset, messageExt);
}
if (startOffsetInfo == null) {
// we should set the check point info to extraInfo field , if the command is popMsg
// find pop ck offset
String key = messageExt.getTopic() + messageExt.getQueueId();
if (!map.containsKey(messageExt.getTopic() + messageExt.getQueueId())) {
map.put(key, ExtraInfoUtil.buildExtraInfo(messageExt.getQueueOffset(), responseHeader.getPopTime(), responseHeader.getInvisibleTime(), responseHeader.getReviveQid(),
messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId()));
}
messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
} else {
String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
int index = sortMap.get(key).indexOf(messageExt.getQueueOffset());
Long msgQueueOffset = msgOffsetInfo.get(key).get(index);
if (msgQueueOffset != messageExt.getQueueOffset()) {
log.warn("Queue offset [{}] of msg is strange, not equal to the stored in msg, {}", msgQueueOffset, messageExt);
}
messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK,
ExtraInfoUtil.buildExtraInfo(startOffsetInfo.get(key), responseHeader.getPopTime(), responseHeader.getInvisibleTime(),
responseHeader.getReviveQid(), messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId(), msgQueueOffset)
);
if (requestHeader.isOrder() && orderCountInfo != null) {
Integer count = orderCountInfo.get(key);
if (count != null && count > 0) {
messageExt.setReconsumeTimes(count);
}
messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK,
ExtraInfoUtil.buildExtraInfo(startOffsetInfo.get(key), responseHeader.getPopTime(), responseHeader.getInvisibleTime(),
responseHeader.getReviveQid(), messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId(), msgQueueOffset)
);
if (requestHeader.isOrder() && orderCountInfo != null) {
Integer count = orderCountInfo.get(key);
if (count != null && count > 0) {
messageExt.setReconsumeTimes(count);
}
}
messageExt.getProperties().computeIfAbsent(MessageConst.PROPERTY_FIRST_POP_TIME, k -> String.valueOf(responseHeader.getPopTime()));
}
messageExt.getProperties().computeIfAbsent(MessageConst.PROPERTY_FIRST_POP_TIME, k -> String.valueOf(responseHeader.getPopTime()));
messageExt.setBrokerName(messageExt.getBrokerName());
messageExt.setTopic(messageQueue.getTopic());
}
}
return popResult;