From 0d516ba708b13aabf206fea2dd184fea1999fdd5 Mon Sep 17 00:00:00 2001 From: fujian-zfj <2573259572@qq.com> Date: Wed, 29 Mar 2023 14:05:07 +0800 Subject: [PATCH] [ISSUE #6473] Fix multi dispatch error when enableMultiDispatch=true and enableLmq=true (#6476) * typo int readme[ecosystem] * remove multi_dispatch to avoid building cq again * if topic starts with %RETRY%, do not process INNER_MULTI_DISPATCH * quickstart consumer test revert --- .../src/main/java/org/apache/rocketmq/store/ConsumeQueue.java | 4 ++-- .../java/org/apache/rocketmq/store/DefaultMessageStore.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java index ed20e39d42..d1c24ee35f 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java +++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java @@ -723,7 +723,7 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle { } private boolean checkMultiDispatchQueue(DispatchRequest dispatchRequest) { - if (!this.messageStore.getMessageStoreConfig().isEnableMultiDispatch()) { + if (!this.messageStore.getMessageStoreConfig().isEnableMultiDispatch() || dispatchRequest.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { return false; } Map prop = dispatchRequest.getPropertiesMap(); @@ -791,7 +791,7 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle { long queueOffset = queueOffsetAssigner.assignQueueOffset(topicQueueKey, messageNum); msg.setQueueOffset(queueOffset); // For LMQ - if (!messageStore.getMessageStoreConfig().isEnableMultiDispatch()) { + if (!messageStore.getMessageStoreConfig().isEnableMultiDispatch() || msg.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { return; } String multiDispatchQueue = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH); diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java index 7798e89b8c..117f204817 100644 --- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java @@ -2812,7 +2812,7 @@ public class DefaultMessageStore implements MessageStore { private void notifyMessageArrive4MultiQueue(DispatchRequest dispatchRequest) { Map prop = dispatchRequest.getPropertiesMap(); - if (prop == null) { + if (prop == null || dispatchRequest.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { return; } String multiDispatchQueue = prop.get(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);