[ISSUE #5965] Fix lmqTopicQueueTable initialization (#5967)

apache · Feb 3, 2023 · fa38abf · fa38abf
1 parent e5637e2
commit fa38abf
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 0 deletions.
diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -19,6 +19,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
+import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -1015,6 +1016,19 @@ public Map<String, Long> getLmqTopicQueueTable() {
         return this.lmqTopicQueueTable;
     }
 
+    public void setLmqTopicQueueTable(Map<String, Long> lmqTopicQueueTable) {
+        if (!defaultMessageStore.getMessageStoreConfig().isEnableLmq()) {
+            return;
+        }
+        Map<String, Long> table = new HashMap<String, Long>(1024);
+        for (Map.Entry<String, Long> entry : lmqTopicQueueTable.entrySet()) {
+            if (MixAll.isLmq(entry.getKey())) {
+                table.put(entry.getKey(), entry.getValue());
+            }
+        }
+        this.lmqTopicQueueTable = table;
+    }
+
     abstract class FlushCommitLogService extends ServiceThread {
         protected static final int RETRY_TIMES_OVER = 10;
     }

diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -1481,6 +1481,7 @@ public void recoverTopicQueueTable() {
         }
 
         this.commitLog.setTopicQueueTable(table);
+        this.commitLog.setLmqTopicQueueTable(table);
     }
 
     public AllocateMappedFileService getAllocateMappedFileService() {