apache
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
Copy file name to clipboard
+11-5Lines changed: 11 additions & 5 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
Copy file name to clipboard
+11-5Lines changed: 11 additions & 5 deletions
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/BrokerPathConfigHelper.java
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/BrokerPathConfigHelper.java
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/LmqPullRequestHoldService.java
Copy file name to clipboard
+62Lines changed: 62 additions & 0 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/LmqPullRequestHoldService.java
Copy file name to clipboard
+62Lines changed: 62 additions & 0 deletions
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/ManyPullRequest.java
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/ManyPullRequest.java
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/PullRequestHoldService.java
Copy file name to clipboard
+4-4Lines changed: 4 additions & 4 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/PullRequestHoldService.java
Copy file name to clipboard
+4-4Lines changed: 4 additions & 4 deletions
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
Copy file name to clipboard
+3-3Lines changed: 3 additions & 3 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
Copy file name to clipboard
+3-3Lines changed: 3 additions & 3 deletions
diff --git a/‎broker/src/main/java/org/apache/rocketmq/broker/offset/LmqConsumerOffsetManager.java
Copy file name to clipboard
+109Lines changed: 109 additions & 0 deletions b/‎broker/src/main/java/org/apache/rocketmq/broker/offset/LmqConsumerOffsetManager.java
Copy file name to clipboard
+109Lines changed: 109 additions & 0 deletions
@@ -47,11 +47,13 @@
 import org.apache.rocketmq.broker.filtersrv.FilterServerManager;
 import org.apache.rocketmq.broker.latency.BrokerFastFailure;
 import org.apache.rocketmq.broker.latency.BrokerFixedThreadPoolExecutor;
+import org.apache.rocketmq.broker.longpolling.LmqPullRequestHoldService;
 import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
 import org.apache.rocketmq.broker.longpolling.PullRequestHoldService;
 import org.apache.rocketmq.broker.mqtrace.ConsumeMessageHook;
 import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.offset.LmqConsumerOffsetManager;
 import org.apache.rocketmq.broker.out.BrokerOuterAPI;
 import org.apache.rocketmq.broker.plugin.MessageStoreFactory;
 import org.apache.rocketmq.broker.plugin.MessageStorePluginContext;
@@ -64,7 +66,9 @@
 import org.apache.rocketmq.broker.processor.ReplyMessageProcessor;
 import org.apache.rocketmq.broker.processor.SendMessageProcessor;
 import org.apache.rocketmq.broker.slave.SlaveSynchronize;
+import org.apache.rocketmq.broker.subscription.LmqSubscriptionGroupManager;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
+import org.apache.rocketmq.broker.topic.LmqTopicConfigManager;
 import org.apache.rocketmq.broker.topic.TopicConfigManager;
 import org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener;
 import org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService;
@@ -106,6 +110,7 @@
 import org.apache.rocketmq.store.dledger.DLedgerCommitLog;
 import org.apache.rocketmq.store.stats.BrokerStats;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
+import org.apache.rocketmq.store.stats.LmqBrokerStatsManager;
 
 public class BrokerController {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -180,18 +185,18 @@ public BrokerController(
         this.nettyServerConfig = nettyServerConfig;
         this.nettyClientConfig = nettyClientConfig;
         this.messageStoreConfig = messageStoreConfig;
-        this.consumerOffsetManager = new ConsumerOffsetManager(this);
-        this.topicConfigManager = new TopicConfigManager(this);
+        this.consumerOffsetManager = messageStoreConfig.isEnableLmq() ? new LmqConsumerOffsetManager(this) : new ConsumerOffsetManager(this);
+        this.topicConfigManager = messageStoreConfig.isEnableLmq() ? new LmqTopicConfigManager(this) : new TopicConfigManager(this);
         this.pullMessageProcessor = new PullMessageProcessor(this);
-        this.pullRequestHoldService = new PullRequestHoldService(this);
+        this.pullRequestHoldService = messageStoreConfig.isEnableLmq() ? new LmqPullRequestHoldService(this) : new PullRequestHoldService(this);
         this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);
         this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
         this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
         this.consumerFilterManager = new ConsumerFilterManager(this);
         this.producerManager = new ProducerManager();
         this.clientHousekeepingService = new ClientHousekeepingService(this);
         this.broker2Client = new Broker2Client(this);
-        this.subscriptionGroupManager = new SubscriptionGroupManager(this);
+        this.subscriptionGroupManager = messageStoreConfig.isEnableLmq() ? new LmqSubscriptionGroupManager(this) : new SubscriptionGroupManager(this);
         this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
         this.filterServerManager = new FilterServerManager(this);
 
@@ -207,7 +212,8 @@ public BrokerController(
         this.heartbeatThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity());
         this.endTransactionThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getEndTransactionPoolQueueCapacity());
 
-        this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
+        this.brokerStatsManager = messageStoreConfig.isEnableLmq() ? new LmqBrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat()) : new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
+
         this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort()));
 
         this.brokerFastFailure = new BrokerFastFailure(this);
 
@@ -39,6 +39,10 @@ public static String getConsumerOffsetPath(final String rootDir) {
         return rootDir + File.separator + "config" + File.separator + "consumerOffset.json";
     }
 
+    public static String getLmqConsumerOffsetPath(final String rootDir) {
+        return rootDir + File.separator + "config" + File.separator + "lmqConsumerOffset.json";
+    }
+
     public static String getSubscriptionGroupPath(final String rootDir) {
         return rootDir + File.separator + "config" + File.separator + "subscriptionGroup.json";
     }
 
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.longpolling;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+
+public class LmqPullRequestHoldService extends PullRequestHoldService {
+    private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+    public LmqPullRequestHoldService(BrokerController brokerController) {
+        super(brokerController);
+    }
+
+    @Override
+    public String getServiceName() {
+        return LmqPullRequestHoldService.class.getSimpleName();
+    }
+
+    @Override
+    public void checkHoldRequest() {
+        for (String key : pullRequestTable.keySet()) {
+            int idx = key.lastIndexOf(TOPIC_QUEUEID_SEPARATOR);
+            if (idx <= 0 || idx >= key.length() - 1) {
+                pullRequestTable.remove(key);
+                continue;
+            }
+            String topic = key.substring(0, idx);
+            int queueId = Integer.parseInt(key.substring(idx + 1));
+            final long offset = brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
+            try {
+                this.notifyMessageArriving(topic, queueId, offset);
+            } catch (Throwable e) {
+                LOGGER.error("check hold request failed. topic={}, queueId={}", topic, queueId, e);
+            }
+            if (MixAll.isLmq(topic)) {
+                ManyPullRequest mpr = pullRequestTable.get(key);
+                if (mpr == null || mpr.getPullRequestList() == null || mpr.getPullRequestList().isEmpty()) {
+                    pullRequestTable.remove(key);
+                }
+            }
+        }
+    }
+}
@@ -39,4 +39,8 @@ public synchronized List<PullRequest> cloneListAndClear() {
 
         return null;
     }
+
+    public ArrayList<PullRequest> getPullRequestList() {
+        return pullRequestList;
+    }
 }
@@ -31,10 +31,10 @@
 
 public class PullRequestHoldService extends ServiceThread {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
-    private static final String TOPIC_QUEUEID_SEPARATOR = "@";
-    private final BrokerController brokerController;
+    protected static final String TOPIC_QUEUEID_SEPARATOR = "@";
+    protected final BrokerController brokerController;
     private final SystemClock systemClock = new SystemClock();
-    private ConcurrentMap<String/* topic@queueId */, ManyPullRequest> pullRequestTable =
+    protected ConcurrentMap<String/* topic@queueId */, ManyPullRequest> pullRequestTable =
         new ConcurrentHashMap<String, ManyPullRequest>(1024);
 
     public PullRequestHoldService(final BrokerController brokerController) {
@@ -93,7 +93,7 @@ public String getServiceName() {
         return PullRequestHoldService.class.getSimpleName();
     }
 
-    private void checkHoldRequest() {
+    protected void checkHoldRequest() {
         for (String key : this.pullRequestTable.keySet()) {
             String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
             if (2 == kArray.length) {
 
@@ -35,12 +35,12 @@
 
 public class ConsumerOffsetManager extends ConfigManager {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
-    private static final String TOPIC_GROUP_SEPARATOR = "@";
+    protected static final String TOPIC_GROUP_SEPARATOR = "@";
 
-    private ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> offsetTable =
+    protected ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> offsetTable =
         new ConcurrentHashMap<String, ConcurrentMap<Integer, Long>>(512);
 
-    private transient BrokerController brokerController;
+    protected transient BrokerController brokerController;
 
     public ConsumerOffsetManager() {
     }
 
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.offset;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.BrokerPathConfigHelper;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class LmqConsumerOffsetManager extends ConsumerOffsetManager {
+    private ConcurrentHashMap<String, Long> lmqOffsetTable = new ConcurrentHashMap<>(512);
+
+    public LmqConsumerOffsetManager(BrokerController brokerController) {
+        super(brokerController);
+    }
+
+    @Override
+    public long queryOffset(final String group, final String topic, final int queueId) {
+        if (!MixAll.isLmq(group)) {
+            return super.queryOffset(group, topic, queueId);
+        }
+        // topic@group
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        Long offset = lmqOffsetTable.get(key);
+        if (offset != null) {
+            return offset;
+        }
+        return -1;
+    }
+
+    @Override
+    public Map<Integer, Long> queryOffset(final String group, final String topic) {
+        if (!MixAll.isLmq(group)) {
+            return super.queryOffset(group, topic);
+        }
+        Map<Integer, Long> map = new HashMap<>();
+        // topic@group
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        Long offset = lmqOffsetTable.get(key);
+        if (offset != null) {
+            map.put(0, offset);
+        }
+        return map;
+    }
+
+    @Override
+    public void commitOffset(final String clientHost, final String group, final String topic, final int queueId,
+        final long offset) {
+        if (!MixAll.isLmq(group)) {
+            super.commitOffset(clientHost, group, topic, queueId, offset);
+            return;
+        }
+        // topic@group
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        lmqOffsetTable.put(key, offset);
+    }
+
+    @Override
+    public String encode() {
+        return this.encode(false);
+    }
+
+    @Override
+    public String configFilePath() {
+        return BrokerPathConfigHelper.getLmqConsumerOffsetPath(brokerController.getMessageStoreConfig().getStorePathRootDir());
+    }
+
+    @Override
+    public void decode(String jsonString) {
+        if (jsonString != null) {
+            LmqConsumerOffsetManager obj = RemotingSerializable.fromJson(jsonString, LmqConsumerOffsetManager.class);
+            if (obj != null) {
+                super.offsetTable = obj.offsetTable;
+                this.lmqOffsetTable = obj.lmqOffsetTable;
+            }
+        }
+    }
+
+    @Override
+    public String encode(final boolean prettyFormat) {
+        return RemotingSerializable.toJson(this, prettyFormat);
+    }
+
+    public ConcurrentHashMap<String, Long> getLmqOffsetTable() {
+        return lmqOffsetTable;
+    }
+
+    public void setLmqOffsetTable(ConcurrentHashMap<String, Long> lmqOffsetTable) {
+        this.lmqOffsetTable = lmqOffsetTable;
+    }
+}
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,10 @@ public static String getConsumerOffsetPath(final String rootDir) {`
`39`	`39`	`return rootDir + File.separator + "config" + File.separator + "consumerOffset.json";`
`40`	`40`	`}`
`41`	`41`
	`42`	`+ public static String getLmqConsumerOffsetPath(final String rootDir) {`
	`43`	`+ return rootDir + File.separator + "config" + File.separator + "lmqConsumerOffset.json";`
	`44`	`+ }`
	`45`	`+`
`42`	`46`	`public static String getSubscriptionGroupPath(final String rootDir) {`
`43`	`47`	`return rootDir + File.separator + "config" + File.separator + "subscriptionGroup.json";`
`44`	`48`	`}`
Original file line number	Diff line number	Diff line change
`@@ -39,4 +39,8 @@ public synchronized List<PullRequest> cloneListAndClear() {`
`39`	`39`
`40`	`40`	`return null;`
`41`	`41`	`}`
	`42`	`+`
	`43`	`+ public ArrayList<PullRequest> getPullRequestList() {`
	`44`	`+ return pullRequestList;`
	`45`	`+ }`
`42`	`46`	`}`