You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "HScarb (via GitHub)" <gi...@apache.org> on 2023/04/11 16:43:28 UTC
[GitHub] [rocketmq] HScarb opened a new pull request, #6577: [ISSUE #6576] Fix pop lmq message
HScarb opened a new pull request, #6577:
URL: https://github.com/apache/rocketmq/pull/6577
**Make sure set the target branch to `develop`**
## What is the purpose of the change
<!--
If this PR fixes a GitHub issue, please add `fixes #<XXX>` or `closes #<XXX>`. Please refer to the documentation for more information:
https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue
-->
fix #6576
## Brief changelog
If use pop to consume a LMQ message, recode message, set LMQ's POP_CK, topic, queueId and queueOffset
## Verifying this change
Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.
- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in [test module](https://github.com/apache/rocketmq/tree/master/test).
- [x] Run `mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle` to make sure basic checks pass. Run `mvn clean install -DskipITs` to make sure unit-test pass. Run `mvn clean test-compile failsafe:integration-test` to make sure integration-test pass.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #6576] Fix pop lmq message [rocketmq]
Posted by "lzkisok (via GitHub)" <gi...@apache.org>.
lzkisok commented on PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#issuecomment-1759568816
目前LMQ支持POP模式消费吗？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1164426234
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java:
##########
@@ -643,14 +643,41 @@ private CompletableFuture<Long> popMsgFromQueue(boolean isRetry, GetMessageResul
atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
String brokerName = brokerController.getBrokerConfig().getBrokerName();
- for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
- // We should not recode buffer for normal topic message
- if (!isRetry) {
- getMessageResult.addMessage(mapedBuffer);
+ for (SelectMappedBufferResult mappedBuffer : result.getMessageMapedList()) {
+ if (MixAll.isLmq(topic)) {
+ List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mappedBuffer.getByteBuffer(),
Review Comment:
Indeed.
I've tried to process it on the client side at first, but we can't get LMQ's queueOffset and queue ID on the client side, because the message was saved with information on the original topic and queue.
So I think it's necessary to process on the broker side and replace the message's info with LMQ (topic, queueOffset and queueId).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1164426234
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java:
##########
@@ -643,14 +643,41 @@ private CompletableFuture<Long> popMsgFromQueue(boolean isRetry, GetMessageResul
atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
String brokerName = brokerController.getBrokerConfig().getBrokerName();
- for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
- // We should not recode buffer for normal topic message
- if (!isRetry) {
- getMessageResult.addMessage(mapedBuffer);
+ for (SelectMappedBufferResult mappedBuffer : result.getMessageMapedList()) {
+ if (MixAll.isLmq(topic)) {
+ List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mappedBuffer.getByteBuffer(),
Review Comment:
Indeed.
I've tried to process it on the client side at first, but we can't get LMQ's queueOffset and queue ID in the client side, because the message was saved with information on the original topic and queue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1218033428
##########
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java:
##########
@@ -1063,73 +1063,124 @@ private PopResult processPopResponse(final String brokerName, final RemotingComm
PopResult popResult = new PopResult(popStatus, msgFoundList);
PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.decodeCommandCustomHeader(PopMessageResponseHeader.class);
popResult.setRestNum(responseHeader.getRestNum());
+ if (popStatus != PopStatus.FOUND) {
+ return popResult;
+ }
// it is a pop command if pop time greater than 0, we should set the check point info to extraInfo field
- if (popStatus == PopStatus.FOUND) {
- Map<String, Long> startOffsetInfo = null;
- Map<String, List<Long>> msgOffsetInfo = null;
- Map<String, Integer> orderCountInfo = null;
+ Map<String, Long> startOffsetInfo = null;
+ Map<String, List<Long>> msgOffsetInfo = null;
+ Map<String, Integer> orderCountInfo = null;
+ if (requestHeader instanceof PopMessageRequestHeader) {
+ 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/*topicMark@queueId*/, List<Long>/*msg queueOffset*/> sortMap
+ = buildQueueOffsetSortedMap(topic, msgFoundList);
+ Map<String, String> map = new HashMap<>(5);
+ for (MessageExt messageExt : msgFoundList) {
if (requestHeader instanceof PopMessageRequestHeader) {
- 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/*topicMark@queueId*/, List<Long>/*msg queueOffset*/> sortMap = new HashMap<>(16);
- for (MessageExt messageExt : msgFoundList) {
- String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
- if (!sortMap.containsKey(key)) {
- sortMap.put(key, new ArrayList<>(4));
- }
- sortMap.get(key).add(messageExt.getQueueOffset());
- }
- Map<String, String> map = new HashMap<>(5);
- for (MessageExt messageExt : msgFoundList) {
- if (requestHeader instanceof PopMessageRequestHeader) {
- 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(), brokerName, messageExt.getQueueId()));
+ 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(), brokerName, messageExt.getQueueId()));
- }
- messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
- } else {
- if (messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
- String queueIdKey = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
- String queueOffsetKey = ExtraInfoUtil.getQueueOffsetMapKey(messageExt.getTopic(), messageExt.getQueueId(), messageExt.getQueueOffset());
- int index = sortMap.get(queueIdKey).indexOf(messageExt.getQueueOffset());
- Long msgQueueOffset = msgOffsetInfo.get(queueIdKey).get(index);
+ }
+ messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
+ } else {
+ if (messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
+ final String queueIdKey;
+ final String queueOffsetKey;
+ final int index;
+ final Long msgQueueOffset;
+ if (MixAll.isLmq(topic) && messageExt.getReconsumeTimes() == 0 && StringUtils.isNotEmpty(
+ messageExt.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH))) {
+ // process LMQ, LMQ topic has only 1 queue, which queue id is 0
Review Comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] pingww commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "pingww (via GitHub)" <gi...@apache.org>.
pingww commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1163763918
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java:
##########
@@ -643,14 +643,41 @@ private CompletableFuture<Long> popMsgFromQueue(boolean isRetry, GetMessageResul
atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
String brokerName = brokerController.getBrokerConfig().getBrokerName();
- for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
- // We should not recode buffer for normal topic message
- if (!isRetry) {
- getMessageResult.addMessage(mapedBuffer);
+ for (SelectMappedBufferResult mappedBuffer : result.getMessageMapedList()) {
+ if (MixAll.isLmq(topic)) {
+ List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mappedBuffer.getByteBuffer(),
Review Comment:
Parsing the message here will affect the performance. Could you consider compatible processing when the client parses the POP response?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#issuecomment-1535881140
@pingww PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] pingww commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "pingww (via GitHub)" <gi...@apache.org>.
pingww commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1166421011
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java:
##########
@@ -643,14 +643,41 @@ private CompletableFuture<Long> popMsgFromQueue(boolean isRetry, GetMessageResul
atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
String brokerName = brokerController.getBrokerConfig().getBrokerName();
- for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
- // We should not recode buffer for normal topic message
- if (!isRetry) {
- getMessageResult.addMessage(mapedBuffer);
+ for (SelectMappedBufferResult mappedBuffer : result.getMessageMapedList()) {
+ if (MixAll.isLmq(topic)) {
+ List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mappedBuffer.getByteBuffer(),
Review Comment:
LMQ's queueOffset is in properties of message, and queueID is default 0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] cserwen commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "cserwen (via GitHub)" <gi...@apache.org>.
cserwen commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1163528087
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java:
##########
@@ -643,14 +643,41 @@ private CompletableFuture<Long> popMsgFromQueue(boolean isRetry, GetMessageResul
atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
String brokerName = brokerController.getBrokerConfig().getBrokerName();
- for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
- // We should not recode buffer for normal topic message
- if (!isRetry) {
- getMessageResult.addMessage(mapedBuffer);
+ for (SelectMappedBufferResult mappedBuffer : result.getMessageMapedList()) {
+ if (MixAll.isLmq(topic)) {
+ List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mappedBuffer.getByteBuffer(),
+ true, false, true);
+ mappedBuffer.release();
+ for (MessageExt messageExt : messageExtList) {
+ try {
+ final Long queueOffset = result.getMessageQueueOffset().get(messageExtList.indexOf(messageExt));
Review Comment:
Traversing the list with index can save the overhead of indexOf.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] codecov-commenter commented on pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#issuecomment-1503788255
## [Codecov](https://codecov.io/gh/apache/rocketmq/pull/6577?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#6577](https://codecov.io/gh/apache/rocketmq/pull/6577?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b4d9cdf) into [develop](https://codecov.io/gh/apache/rocketmq/commit/f44a1c3fd5e218f1038b360245479de54f83b408?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (f44a1c3) will **increase** coverage by `0.03%`.
> The diff coverage is `77.77%`.
```diff
@@ Coverage Diff @@
## develop #6577 +/- ##
=============================================
+ Coverage 43.08% 43.11% +0.03%
- Complexity 8994 8998 +4
=============================================
Files 1107 1107
Lines 78257 78278 +21
Branches 10201 10203 +2
=============================================
+ Hits 33716 33750 +34
+ Misses 40318 40305 -13
Partials 4223 4223
```
| [Impacted Files](https://codecov.io/gh/apache/rocketmq/pull/6577?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...rocketmq/broker/processor/PopMessageProcessor.java](https://codecov.io/gh/apache/rocketmq/pull/6577?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvcHJvY2Vzc29yL1BvcE1lc3NhZ2VQcm9jZXNzb3IuamF2YQ==) | `39.67% <77.77%> (+2.06%)` | :arrow_up: |
... and [16 files with indirect coverage changes](https://codecov.io/gh/apache/rocketmq/pull/6577/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] pingww merged pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "pingww (via GitHub)" <gi...@apache.org>.
pingww merged PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] pingww commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "pingww (via GitHub)" <gi...@apache.org>.
pingww commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1208768081
##########
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java:
##########
@@ -1063,73 +1063,124 @@ private PopResult processPopResponse(final String brokerName, final RemotingComm
PopResult popResult = new PopResult(popStatus, msgFoundList);
PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.decodeCommandCustomHeader(PopMessageResponseHeader.class);
popResult.setRestNum(responseHeader.getRestNum());
+ if (popStatus != PopStatus.FOUND) {
+ return popResult;
+ }
// it is a pop command if pop time greater than 0, we should set the check point info to extraInfo field
- if (popStatus == PopStatus.FOUND) {
- Map<String, Long> startOffsetInfo = null;
- Map<String, List<Long>> msgOffsetInfo = null;
- Map<String, Integer> orderCountInfo = null;
+ Map<String, Long> startOffsetInfo = null;
+ Map<String, List<Long>> msgOffsetInfo = null;
+ Map<String, Integer> orderCountInfo = null;
+ if (requestHeader instanceof PopMessageRequestHeader) {
+ 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/*topicMark@queueId*/, List<Long>/*msg queueOffset*/> sortMap
+ = buildQueueOffsetSortedMap(topic, msgFoundList);
+ Map<String, String> map = new HashMap<>(5);
+ for (MessageExt messageExt : msgFoundList) {
if (requestHeader instanceof PopMessageRequestHeader) {
- 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/*topicMark@queueId*/, List<Long>/*msg queueOffset*/> sortMap = new HashMap<>(16);
- for (MessageExt messageExt : msgFoundList) {
- String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
- if (!sortMap.containsKey(key)) {
- sortMap.put(key, new ArrayList<>(4));
- }
- sortMap.get(key).add(messageExt.getQueueOffset());
- }
- Map<String, String> map = new HashMap<>(5);
- for (MessageExt messageExt : msgFoundList) {
- if (requestHeader instanceof PopMessageRequestHeader) {
- 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(), brokerName, messageExt.getQueueId()));
+ 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(), brokerName, messageExt.getQueueId()));
- }
- messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
- } else {
- if (messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
- String queueIdKey = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
- String queueOffsetKey = ExtraInfoUtil.getQueueOffsetMapKey(messageExt.getTopic(), messageExt.getQueueId(), messageExt.getQueueOffset());
- int index = sortMap.get(queueIdKey).indexOf(messageExt.getQueueOffset());
- Long msgQueueOffset = msgOffsetInfo.get(queueIdKey).get(index);
+ }
+ messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
+ } else {
+ if (messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
+ final String queueIdKey;
+ final String queueOffsetKey;
+ final int index;
+ final Long msgQueueOffset;
+ if (MixAll.isLmq(topic) && messageExt.getReconsumeTimes() == 0 && StringUtils.isNotEmpty(
+ messageExt.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH))) {
+ // process LMQ, LMQ topic has only 1 queue, which queue id is 0
Review Comment:
Should we define this queue id 0 as a constant ？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] vongosling commented on a diff in pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "vongosling (via GitHub)" <gi...@apache.org>.
vongosling commented on code in PR #6577:
URL: https://github.com/apache/rocketmq/pull/6577#discussion_r1174524348
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java:
##########
@@ -643,14 +643,41 @@ private CompletableFuture<Long> popMsgFromQueue(boolean isRetry, GetMessageResul
atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
String brokerName = brokerController.getBrokerConfig().getBrokerName();
- for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
- // We should not recode buffer for normal topic message
- if (!isRetry) {
- getMessageResult.addMessage(mapedBuffer);
+ for (SelectMappedBufferResult mappedBuffer : result.getMessageMapedList()) {
+ if (MixAll.isLmq(topic)) {
+ List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mappedBuffer.getByteBuffer(),
Review Comment:
Take another case, if we hope to use the older client to visit our new broker version. Does it enough to modify in the client side?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb closed pull request #6577: [ISSUE #6576] Fix pop lmq message
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb closed pull request #6577: [ISSUE #6576] Fix pop lmq message
URL: https://github.com/apache/rocketmq/pull/6577
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org