You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2023/06/02 02:36:16 UTC
[rocketmq] branch develop updated: [ISSUE #6779] Support scheduled message for LMQ (#6780)
This is an automated email from the ASF dual-hosted git repository.
jinrongtong pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 1d62b74ab6 [ISSUE #6779] Support scheduled message for LMQ (#6780)
1d62b74ab6 is described below
commit 1d62b74ab6ee1e2477208d9a15a8c21f121ecbab
Author: cnScarb <jj...@163.com>
AuthorDate: Fri Jun 2 10:36:09 2023 +0800
[ISSUE #6779] Support scheduled message for LMQ (#6780)
---
.../main/java/org/apache/rocketmq/store/ConsumeQueue.java | 12 ++++++++++--
1 file changed, 10 insertions(+), 2 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 695f0e69fb..0c44ad043f 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
@@ -30,6 +30,7 @@ import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
+import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
@@ -40,6 +41,7 @@ import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.FileQueueLifeCycle;
import org.apache.rocketmq.store.queue.QueueOffsetOperator;
import org.apache.rocketmq.store.queue.ReferredIterator;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
@@ -722,7 +724,10 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
}
private boolean checkMultiDispatchQueue(DispatchRequest dispatchRequest) {
- if (!this.messageStore.getMessageStoreConfig().isEnableMultiDispatch() || dispatchRequest.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
+ if (!this.messageStore.getMessageStoreConfig().isEnableMultiDispatch()
+ || dispatchRequest.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
+ || dispatchRequest.getTopic().equals(TimerMessageStore.TIMER_TOPIC)
+ || dispatchRequest.getTopic().equals(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC)) {
return false;
}
Map<String, String> prop = dispatchRequest.getPropertiesMap();
@@ -840,7 +845,10 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
}
public boolean isNeedHandleMultiDispatch(MessageExtBrokerInner msg) {
- return messageStore.getMessageStoreConfig().isEnableMultiDispatch() && !msg.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX);
+ return messageStore.getMessageStoreConfig().isEnableMultiDispatch()
+ && !msg.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
+ && !msg.getTopic().equals(TimerMessageStore.TIMER_TOPIC)
+ && !msg.getTopic().equals(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC);
}
public String queueKey(String queueName, MessageExtBrokerInner msgInner) {