[ISSUE #6545] Remove getTopicConfigs method in interface MessageStore (#6531)

* change map -> lambda

* f

* fix unit test

* remove getTopicConfig function

* Update MultiDispatchTest.java

* Update CompactionStore.java

* update

* update test

* update test

* Update BatchConsumeMessageTest.java

* Update BrokerController.java

* Update BrokerController.java

* check

* Update BrokerController.java

* Update BatchConsumeMessageTest.java

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -737,8 +737,7 @@ public class BrokerController {
 
         if (result) {
             try {
-                DefaultMessageStore defaultMessageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig);
-                defaultMessageStore.setTopicConfigTable(topicConfigManager.getTopicConfigTable());
+                DefaultMessageStore defaultMessageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
 
                 if (messageStoreConfig.isEnableDLegerCommitLog()) {
                     DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, defaultMessageStore);

broker/src/test/java/org/apache/rocketmq/broker/filter/MessageStoreWithFilterTest.java

@@ -26,6 +26,7 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
@@ -159,7 +160,7 @@ public class MessageStoreWithFilterTest {
                                      long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
                 }
             }
-            , brokerConfig);
+            , brokerConfig, new ConcurrentHashMap<>());
 
         master.getDispatcherList().addFirst(new CommitLogDispatcher() {
             @Override

broker/src/test/java/org/apache/rocketmq/broker/schedule/ScheduleMessageServiceTest.java

@@ -119,7 +119,7 @@ public class ScheduleMessageServiceTest {
 
         brokerConfig = new BrokerConfig();
         BrokerStatsManager manager = new BrokerStatsManager(brokerConfig.getBrokerClusterName(), brokerConfig.isEnableDetailStat());
-        messageStore = new DefaultMessageStore(messageStoreConfig, manager, new MyMessageArrivingListener(), new BrokerConfig());
+        messageStore = new DefaultMessageStore(messageStoreConfig, manager, new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
 
         assertThat(messageStore.load()).isTrue();
 

store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

@@ -200,16 +200,20 @@ public class DefaultMessageStore implements MessageStore {
 
     private long stateMachineVersion = 0L;
 
+    // this is a unmodifiableMap
+    private ConcurrentMap<String, TopicConfig> topicConfigTable;
+
     private final ScheduledExecutorService scheduledCleanQueueExecutorService =
         Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreCleanQueueScheduledThread"));
 
     public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
-        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
+        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig, final ConcurrentMap<String, TopicConfig> topicConfigTable) throws IOException {
         this.messageArrivingListener = messageArrivingListener;
         this.brokerConfig = brokerConfig;
         this.messageStoreConfig = messageStoreConfig;
         this.aliveReplicasNum = messageStoreConfig.getTotalReplicas();
         this.brokerStatsManager = brokerStatsManager;
+        this.topicConfigTable = topicConfigTable;
         this.allocateMappedFileService = new AllocateMappedFileService(this);
         if (messageStoreConfig.isEnableDLegerCommitLog()) {
             this.commitLog = new DLedgerCommitLog(this);
@@ -2047,18 +2051,16 @@ public class DefaultMessageStore implements MessageStore {
         }
     }
 
-    @Override
     public ConcurrentMap<String, TopicConfig> getTopicConfigs() {
-        return this.consumeQueueStore.getTopicConfigs();
+        return this.topicConfigTable;
     }
 
-    @Override
     public Optional<TopicConfig> getTopicConfig(String topic) {
-        return this.consumeQueueStore.getTopicConfig(topic);
-    }
+        if (this.topicConfigTable == null) {
+            return Optional.empty();
+        }
 
-    public void setTopicConfigTable(ConcurrentMap<String, TopicConfig> topicConfigTable) {
-        this.consumeQueueStore.setTopicConfigTable(topicConfigTable);
+        return Optional.ofNullable(this.topicConfigTable.get(topic));
     }
 
     public BrokerIdentity getBrokerIdentity() {

store/src/main/java/org/apache/rocketmq/store/MessageStore.java

@@ -25,15 +25,12 @@ import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.SystemClock;
-import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageExtBatch;
 import org.apache.rocketmq.common.message.MessageExtBrokerInner;
@@ -736,21 +733,6 @@ public interface MessageStore {
      */
     void assignOffset(MessageExtBrokerInner msg, short messageNum);
 
-    /**
-     * get all topic config
-     *
-     * @return all topic config info
-     */
-    Map<String, TopicConfig> getTopicConfigs();
-
-    /**
-     * get topic config
-     *
-     * @param topic topic name
-     * @return topic config info
-     */
-    Optional<TopicConfig> getTopicConfig(String topic);
-
     /**
      * Get master broker message store in process in broker container
      *

store/src/main/java/org/apache/rocketmq/store/kv/CompactionStore.java

@@ -29,9 +29,9 @@ import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.utils.CleanupPolicyUtils;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.DefaultMessageStore;
 import org.apache.rocketmq.store.DispatchRequest;
 import org.apache.rocketmq.store.GetMessageResult;
-import org.apache.rocketmq.store.MessageStore;
 import org.apache.rocketmq.store.SelectMappedBufferResult;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 
@@ -53,7 +53,7 @@ public class CompactionStore {
     private final String compactionPath;
     private final String compactionLogPath;
     private final String compactionCqPath;
-    private final MessageStore defaultMessageStore;
+    private final DefaultMessageStore defaultMessageStore;
     private final CompactionPositionMgr positionMgr;
     private final ConcurrentHashMap<String, CompactionLog> compactionLogTable;
     private final ScheduledExecutorService compactionSchedule;
@@ -65,7 +65,7 @@ public class CompactionStore {
 
     private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
 
-    public CompactionStore(MessageStore defaultMessageStore) {
+    public CompactionStore(DefaultMessageStore defaultMessageStore) {
         this.defaultMessageStore = defaultMessageStore;
         this.compactionLogTable = new ConcurrentHashMap<>();
         MessageStoreConfig config = defaultMessageStore.getMessageStoreConfig();

store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java

@@ -26,15 +26,12 @@ import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.SystemClock;
-import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageExtBatch;
 import org.apache.rocketmq.common.message.MessageExtBrokerInner;
@@ -595,16 +592,6 @@ public abstract class AbstractPluginMessageStore implements MessageStore {
         next.assignOffset(msg, messageNum);
     }
 
-    @Override
-    public Map<String, TopicConfig> getTopicConfigs() {
-        return next.getTopicConfigs();
-    }
-
-    @Override
-    public Optional<TopicConfig> getTopicConfig(String topic) {
-        return next.getTopicConfig(topic);
-    }
-
     @Override
     public List<PutMessageHook> getPutMessageHookList() {
         return next.getPutMessageHookList();

store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStore.java

@@ -64,20 +64,12 @@ public class ConsumeQueueStore {
     protected final QueueOffsetAssigner queueOffsetAssigner = new QueueOffsetAssigner();
     protected final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface>> consumeQueueTable;
 
-    // Should be careful, do not change the topic config
-    // TopicConfigManager is more suitable here.
-    private ConcurrentMap<String, TopicConfig> topicConfigTable;
-
     public ConsumeQueueStore(DefaultMessageStore messageStore, MessageStoreConfig messageStoreConfig) {
         this.messageStore = messageStore;
         this.messageStoreConfig = messageStoreConfig;
         this.consumeQueueTable = new ConcurrentHashMap<>(32);
     }
 
-    public void setTopicConfigTable(ConcurrentMap<String, TopicConfig> topicConfigTable) {
-        this.topicConfigTable = topicConfigTable;
-    }
-
     private FileQueueLifeCycle getLifeCycle(String topic, int queueId) {
         return findOrCreateConsumeQueue(topic, queueId);
     }
@@ -173,9 +165,9 @@ public class ConsumeQueueStore {
     }
 
     private void queueTypeShouldBe(String topic, CQType cqTypeExpected) {
-        TopicConfig topicConfig = this.topicConfigTable == null ? null : this.topicConfigTable.get(topic);
+        Optional<TopicConfig> topicConfig = this.messageStore.getTopicConfig(topic);
 
-        CQType cqTypeActual = QueueTypeUtils.getCQType(Optional.ofNullable(topicConfig));
+        CQType cqTypeActual = QueueTypeUtils.getCQType(topicConfig);
 
         if (!Objects.equals(cqTypeExpected, cqTypeActual)) {
             throw new RuntimeException(format("The queue type of topic: %s should be %s, but is %s", topic, cqTypeExpected, cqTypeActual));
@@ -341,7 +333,7 @@ public class ConsumeQueueStore {
 
         ConsumeQueueInterface newLogic;
 
-        Optional<TopicConfig> topicConfig = this.getTopicConfig(topic);
+        Optional<TopicConfig> topicConfig = this.messageStore.getTopicConfig(topic);
         // TODO maybe the topic has been deleted.
         if (Objects.equals(CQType.BatchCQ, QueueTypeUtils.getCQType(topicConfig))) {
             newLogic = new BatchConsumeQueue(
@@ -537,18 +529,6 @@ public class ConsumeQueueStore {
         }
     }
 
-    public ConcurrentMap<String, TopicConfig> getTopicConfigs() {
-        return this.topicConfigTable;
-    }
-
-    public Optional<TopicConfig> getTopicConfig(String topic) {
-        if (this.topicConfigTable == null) {
-            return Optional.empty();
-        }
-
-        return Optional.ofNullable(this.topicConfigTable.get(topic));
-    }
-
     public long getTotalSize() {
         long totalSize = 0;
         for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {

store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java

@@ -25,6 +25,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.message.Message;
@@ -55,7 +56,7 @@ public class AppendCallbackTest {
         messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore");
         messageStoreConfig.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore" + File.separator + "commitlog");
         //too much reference
-        DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, new BrokerConfig());
+        DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, new BrokerConfig(), new ConcurrentHashMap<>());
         CommitLog commitLog = new CommitLog(messageStore);
         callback = commitLog.new DefaultAppendMessageCallback();
     }

store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java

@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.message.Message;
@@ -78,7 +79,7 @@ public class BatchPutMessageTest {
         messageStoreConfig.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + File.separator
             + "putmessagesteststore" + File.separator + "commitlog");
         messageStoreConfig.setHaListenPort(0);
-        return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), new MyMessageArrivingListener(), new BrokerConfig());
+        return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
     }
 
     @Test

store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java

@@ -24,6 +24,7 @@ import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.net.UnknownHostException;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 
@@ -151,7 +152,7 @@ public class ConsumeQueueTest {
                     long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
                 }
             }
-            , brokerConfig);
+            , brokerConfig, new ConcurrentHashMap<>());
 
         assertThat(master.load()).isTrue();
 
@@ -179,7 +180,7 @@ public class ConsumeQueueTest {
                     long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
                 }
             }
-            , brokerConfig);
+            , brokerConfig, new ConcurrentHashMap<>());
 
         assertThat(master.load()).isTrue();
 

store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreCleanFilesTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.store;
 
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.UtilAll;
@@ -483,7 +484,7 @@ public class DefaultMessageStoreCleanFilesTest {
 
     private void initMessageStore(MessageStoreConfig messageStoreConfig, double diskSpaceCleanForciblyRatio) throws Exception {
         messageStore = new DefaultMessageStore(messageStoreConfig,
-                new BrokerStatsManager("test", true), new MyMessageArrivingListener(), new BrokerConfig());
+                new BrokerStatsManager("test", true), new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
 
         cleanCommitLogService = getCleanCommitLogService();
         cleanConsumeQueueService = getCleanConsumeQueueService();

store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreShutDownTest.java

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.store;
 
 import java.io.File;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.store.config.FlushDiskType;
@@ -74,7 +75,7 @@ public class DefaultMessageStoreShutDownTest {
         String storeRootPath = System.getProperty("java.io.tmpdir") + File.separator + "store";
         messageStoreConfig.setStorePathRootDir(storeRootPath);
         messageStoreConfig.setHaListenPort(0);
-        return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), null, new BrokerConfig());
+        return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), null, new BrokerConfig(), new ConcurrentHashMap<>());
     }
 
 }

store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java

@@ -99,7 +99,7 @@ public class DefaultMessageStoreTest {
         messageStoreConfig.setMaxIndexNum(100 * 10);
         messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "store");
         messageStoreConfig.setHaListenPort(0);
-        MessageStore master = new DefaultMessageStore(messageStoreConfig, null, new MyMessageArrivingListener(), new BrokerConfig());
+        MessageStore master = new DefaultMessageStore(messageStoreConfig, null, new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
 
         boolean load = master.load();
         assertTrue(load);
@@ -144,7 +144,7 @@ public class DefaultMessageStoreTest {
         return new DefaultMessageStore(messageStoreConfig,
             new BrokerStatsManager("simpleTest", true),
             new MyMessageArrivingListener(),
-            new BrokerConfig());
+            new BrokerConfig(), new ConcurrentHashMap<>());
     }
 
     @Test

store/src/test/java/org/apache/rocketmq/store/HATest.java

@@ -27,6 +27,7 @@ import java.net.SocketAddress;
 import java.time.Duration;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
@@ -250,7 +251,7 @@ public class HATest {
     private MessageStore buildMessageStore(MessageStoreConfig messageStoreConfig, long brokerId) throws Exception {
         BrokerConfig brokerConfig = new BrokerConfig();
         brokerConfig.setBrokerId(brokerId);
-        return new DefaultMessageStore(messageStoreConfig, brokerStatsManager, null, brokerConfig);
+        return new DefaultMessageStore(messageStoreConfig, brokerStatsManager, null, brokerConfig, new ConcurrentHashMap<>());
     }
 
     private void buildMessageStoreConfig(MessageStoreConfig messageStoreConfig) {

store/src/test/java/org/apache/rocketmq/store/MultiDispatchTest.java

@@ -21,6 +21,7 @@ import java.io.File;
 import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
 
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.message.MessageConst;
@@ -57,7 +58,7 @@ public class MultiDispatchTest {
         messageStoreConfig.setEnableMultiDispatch(true);
         BrokerConfig brokerConfig = new BrokerConfig();
         //too much reference
-        messageStore = new DefaultMessageStore(messageStoreConfig, null, null, brokerConfig);
+        messageStore = new DefaultMessageStore(messageStoreConfig, null, null, brokerConfig, new ConcurrentHashMap<>());
         consumeQueue = new ConsumeQueue("xxx", 0,
             getStorePathConsumeQueue(messageStoreConfig.getStorePathRootDir()), messageStoreConfig.getMappedFileSizeConsumeQueue(), messageStore);
     }

store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerMultiPathTest.java

@@ -21,6 +21,7 @@ import java.io.File;
 import java.time.Duration;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.store.DefaultMessageStore;
@@ -107,7 +108,7 @@ public class DLedgerMultiPathTest extends MessageStoreTestBase {
         storeConfig.setdLegerSelfId(selfId);
         DefaultMessageStore defaultMessageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("DLedgerCommitLogTest", true), (topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {
 
-        }, new BrokerConfig());
+        }, new BrokerConfig(), new ConcurrentHashMap<>());
         Assert.assertTrue(defaultMessageStore.load());
         defaultMessageStore.start();
         return defaultMessageStore;

store/src/test/java/org/apache/rocketmq/store/dledger/MessageStoreTestBase.java

@@ -21,6 +21,7 @@ import io.openmessaging.storage.dledger.DLedgerServer;
 import java.io.File;
 import java.net.UnknownHostException;
 import java.util.Arrays;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
@@ -60,7 +61,7 @@ public class MessageStoreTestBase extends StoreTestBase {
         storeConfig.setRecheckReputOffsetFromCq(true);
         DefaultMessageStore defaultMessageStore = new DefaultMessageStore(storeConfig,  new BrokerStatsManager("DLedgerCommitlogTest", true), (topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {
 
-        }, new BrokerConfig());
+        }, new BrokerConfig(), new ConcurrentHashMap<>());
         DLedgerServer dLegerServer = ((DLedgerCommitLog) defaultMessageStore.getCommitLog()).getdLedgerServer();
         if (leaderId != null) {
             dLegerServer.getdLedgerConfig().setEnableLeaderElector(false);
@@ -109,7 +110,7 @@ public class MessageStoreTestBase extends StoreTestBase {
         storeConfig.setFlushDiskType(FlushDiskType.ASYNC_FLUSH);
         DefaultMessageStore defaultMessageStore = new DefaultMessageStore(storeConfig,  new BrokerStatsManager("CommitlogTest", true), (topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {
 
-        }, new BrokerConfig());
+        }, new BrokerConfig(), new ConcurrentHashMap<>());
 
         if (createAbort) {
             String fileName = StorePathConfigHelper.getAbortFile(storeConfig.getStorePathRootDir());

store/src/test/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHATest.java

@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.store.ha.autoswitch;
 
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
@@ -497,7 +498,7 @@ public class AutoSwitchHATest {
         BrokerConfig brokerConfig = new BrokerConfig();
         brokerConfig.setBrokerId(brokerId);
         brokerConfig.setEnableControllerMode(true);
-        return new DefaultMessageStore(messageStoreConfig, brokerStatsManager, null, brokerConfig);
+        return new DefaultMessageStore(messageStoreConfig, brokerStatsManager, null, brokerConfig, new ConcurrentHashMap<>());
     }
 
     private void buildMessageStoreConfig(MessageStoreConfig messageStoreConfig, int mappedFileSize) {

store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java

@@ -27,6 +27,9 @@ import java.util.Objects;
 import java.util.Queue;
 import java.util.Random;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.attribute.CQType;
 import org.apache.rocketmq.common.message.MessageAccessor;
@@ -37,10 +40,10 @@ import org.apache.rocketmq.common.message.MessageExtBrokerInner;
 import org.apache.rocketmq.common.sysflag.MessageSysFlag;
 import org.apache.rocketmq.common.utils.QueueTypeUtils;
 import org.apache.rocketmq.store.ConsumeQueueExt;
+import org.apache.rocketmq.store.DefaultMessageStore;
 import org.apache.rocketmq.store.GetMessageResult;
 import org.apache.rocketmq.store.GetMessageStatus;
 import org.apache.rocketmq.store.MessageFilter;
-import org.apache.rocketmq.store.MessageStore;
 import org.apache.rocketmq.store.PutMessageResult;
 import org.apache.rocketmq.store.PutMessageStatus;
 import org.apache.rocketmq.store.SelectMappedBufferResult;
@@ -55,11 +58,13 @@ import static org.awaitility.Awaitility.await;
 public class BatchConsumeMessageTest extends QueueTestBase {
     private static final int BATCH_NUM = 10;
     private static final int TOTAL_MSGS = 200;
-    private MessageStore messageStore;
+    private DefaultMessageStore messageStore;
+    private ConcurrentMap<String, TopicConfig> topicConfigTableMap;
 
     @Before
     public void init() throws Exception {
-        messageStore = createMessageStore(null, true);
+        this.topicConfigTableMap = new ConcurrentHashMap<>();
+        messageStore = (DefaultMessageStore) createMessageStore(null, true, this.topicConfigTableMap);
         messageStore.load();
         messageStore.start();
     }
@@ -76,7 +81,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     @Test
     public void testSendMessagesToCqTopic() {
         String topic = UUID.randomUUID().toString();
-        createTopic(topic, CQType.SimpleCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.SimpleCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
 //        int batchNum = 10;
 
@@ -100,7 +106,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     @Test
     public void testSendMessagesToBcqTopic() {
         String topic = UUID.randomUUID().toString();
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         // case 1 has PROPERTY_INNER_NUM but has no INNER_BATCH_FLAG
 //        MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, 1);
@@ -123,7 +130,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     @Test
     public void testConsumeBatchMessage() {
         String topic = UUID.randomUUID().toString();
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
         int batchNum = 10;
 
         MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, batchNum);
@@ -153,7 +161,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     @Test
     public void testNextBeginOffsetConsumeBatchMessage() {
         String topic = UUID.randomUUID().toString();
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
         Random random = new Random();
         int putMessageCount = 1000;
 
@@ -191,7 +200,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     public void testGetOffsetInQueueByTime() throws Exception {
         String topic = "testGetOffsetInQueueByTime";
 
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
         Assert.assertTrue(QueueTypeUtils.isBatchCq(messageStore.getTopicConfig(topic)));
 
         // The initial min max offset, before and after the creation of consume queue
@@ -231,7 +241,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     @Test
     public void testDispatchNormalConsumeQueue() throws Exception {
         String topic = "TestDispatchBuildConsumeQueue";
-        createTopic(topic, CQType.SimpleCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.SimpleCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         long timeStart = -1;
         long timeMid = -1;
@@ -291,7 +302,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
         long timeStart = -1;
         long timeMid = -1;
 
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         for (int i = 0; i < 100; i++) {
             PutMessageResult putMessageResult = messageStore.putMessage(buildMessage(topic, batchNum));
@@ -349,7 +361,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     public void testGetBatchMessageWithinNumber() {
         String topic = UUID.randomUUID().toString();
 
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         int batchNum = 20;
         for (int i = 0; i < 200; i++) {
@@ -409,7 +422,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     @Test
     public void testGetBatchMessageWithinSize() {
         String topic = UUID.randomUUID().toString();
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         int batchNum = 10;
         for (int i = 0; i < 100; i++) {
@@ -463,7 +477,8 @@ public class BatchConsumeMessageTest extends QueueTestBase {
     }
 
     protected void putMsg(String topic) {
-        createTopic(topic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(topic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         for (int i = 0; i < TOTAL_MSGS; i++) {
             MessageExtBrokerInner message = buildMessage(topic, BATCH_NUM * (i % 2 + 1));

store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.store.queue;
 
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.store.ConsumeQueue;
 import org.apache.rocketmq.store.DefaultMessageStore;
@@ -299,7 +300,7 @@ public class BatchConsumeQueueTest extends StoreTestBase {
             new BrokerStatsManager("simpleTest", true),
             (topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {
             },
-            new BrokerConfig());
+            new BrokerConfig(), new ConcurrentHashMap<>());
     }
 
 }

store/src/test/java/org/apache/rocketmq/store/queue/ConsumeQueueStoreTest.java

@@ -16,9 +16,11 @@
  */
 package org.apache.rocketmq.store.queue;
 
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.attribute.CQType;
-import org.apache.rocketmq.store.DefaultMessageStore;
 import org.apache.rocketmq.store.MessageStore;
 import org.apache.rocketmq.store.PutMessageResult;
 import org.apache.rocketmq.store.PutMessageStatus;
@@ -36,10 +38,14 @@ import static org.junit.Assert.assertEquals;
 
 public class ConsumeQueueStoreTest extends QueueTestBase {
     private MessageStore messageStore;
+    private ConcurrentMap<String, TopicConfig> topicConfigTableMap;
+
+
 
     @Before
     public void init() throws Exception {
-        messageStore = createMessageStore(null, true);
+        this.topicConfigTableMap = new ConcurrentHashMap<>();
+        messageStore = createMessageStore(null, true, topicConfigTableMap);
         messageStore.load();
         messageStore.start();
     }
@@ -56,7 +62,8 @@ public class ConsumeQueueStoreTest extends QueueTestBase {
     @Test
     public void testLoadConsumeQueuesWithWrongAttribute() {
         String normalTopic = UUID.randomUUID().toString();
-        createTopic(normalTopic, CQType.SimpleCQ, messageStore);
+        ConcurrentMap<String, TopicConfig> topicConfigTable = createTopicConfigTable(normalTopic, CQType.SimpleCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         for (int i = 0; i < 10; i++) {
             PutMessageResult putMessageResult = messageStore.putMessage(buildMessage(normalTopic, -1));
@@ -66,10 +73,10 @@ public class ConsumeQueueStoreTest extends QueueTestBase {
         await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
 
         // simulate delete topic but with files left.
-        ((DefaultMessageStore)messageStore).setTopicConfigTable(null);
+        this.topicConfigTableMap.clear();
 
-        createTopic(normalTopic, CQType.BatchCQ, messageStore);
-        messageStore.shutdown();
+        topicConfigTable = createTopicConfigTable(normalTopic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         RuntimeException runtimeException = Assert.assertThrows(RuntimeException.class, () -> messageStore.getQueueStore().load());
         Assert.assertTrue(runtimeException.getMessage().endsWith("should be SimpleCQ, but is BatchCQ"));
@@ -78,7 +85,8 @@ public class ConsumeQueueStoreTest extends QueueTestBase {
     @Test
     public void testLoadBatchConsumeQueuesWithWrongAttribute() {
         String batchTopic = UUID.randomUUID().toString();
-        createTopic(batchTopic, CQType.BatchCQ, messageStore);
+        ConcurrentMap<String, TopicConfig>  topicConfigTable = createTopicConfigTable(batchTopic, CQType.BatchCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
 
         for (int i = 0; i < 10; i++) {
             PutMessageResult putMessageResult = messageStore.putMessage(buildMessage(batchTopic, 10));
@@ -88,9 +96,10 @@ public class ConsumeQueueStoreTest extends QueueTestBase {
         await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
 
         // simulate delete topic but with files left.
-        ((DefaultMessageStore)messageStore).setTopicConfigTable(null);
+        this.topicConfigTableMap.clear();
 
-        createTopic(batchTopic, CQType.SimpleCQ, messageStore);
+        topicConfigTable = createTopicConfigTable(batchTopic, CQType.SimpleCQ);
+        this.topicConfigTableMap.putAll(topicConfigTable);
         messageStore.shutdown();
 
         RuntimeException runtimeException = Assert.assertThrows(RuntimeException.class, () -> messageStore.getQueueStore().load());

store/src/test/java/org/apache/rocketmq/store/queue/ConsumeQueueTest.java

@@ -20,6 +20,7 @@ import java.io.File;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.attribute.CQType;
@@ -74,7 +75,7 @@ public class ConsumeQueueTest extends QueueTestBase {
         DefaultMessageStore master = new DefaultMessageStore(
             messageStoreConfig, new BrokerStatsManager(brokerConfig),
             (topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {
-            }, brokerConfig);
+            }, brokerConfig, new ConcurrentHashMap<>());
 
         assertThat(master.load()).isTrue();
 
@@ -112,7 +113,7 @@ public class ConsumeQueueTest extends QueueTestBase {
     public void testIterator() throws Exception {
         final int msgNum = 100;
         final int msgSize = 1000;
-        MessageStore messageStore =  createMessageStore(null, true);
+        MessageStore messageStore =  createMessageStore(null, true, null);
         messageStore.load();
         String topic = UUID.randomUUID().toString();
         //The initial min max offset, before and after the creation of consume queue

store/src/test/java/org/apache/rocketmq/store/queue/QueueTestBase.java

@@ -41,7 +41,7 @@ import java.util.concurrent.ConcurrentMap;
 
 public class QueueTestBase extends StoreTestBase {
 
-    protected void createTopic(String topic, CQType cqType, MessageStore messageStore) {
+    protected ConcurrentMap<String, TopicConfig> createTopicConfigTable(String topic, CQType cqType) {
         ConcurrentMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
         TopicConfig topicConfigToBeAdded = new TopicConfig();
 
@@ -51,14 +51,14 @@ public class QueueTestBase extends StoreTestBase {
         topicConfigToBeAdded.setAttributes(attributes);
 
         topicConfigTable.put(topic, topicConfigToBeAdded);
-        ((DefaultMessageStore) messageStore).setTopicConfigTable(topicConfigTable);
+        return topicConfigTable;
     }
 
     protected Callable<Boolean> fullyDispatched(MessageStore messageStore) {
         return () -> messageStore.dispatchBehindBytes() == 0;
     }
 
-    protected MessageStore createMessageStore(String baseDir, boolean extent) throws Exception {
+    protected MessageStore createMessageStore(String baseDir, boolean extent,  ConcurrentMap<String, TopicConfig> topicConfigTable) throws Exception {
         if (baseDir == null) {
             baseDir = createBaseDir();
         }
@@ -86,7 +86,7 @@ public class QueueTestBase extends StoreTestBase {
             new BrokerStatsManager("simpleTest", true),
             (topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {
             },
-            new BrokerConfig());
+            new BrokerConfig(), topicConfigTable);
     }
 
     public MessageExtBrokerInner buildMessage(String topic, int batchNum) {

store/src/test/java/org/apache/rocketmq/store/timer/TimerMessageStoreTest.java

@@ -29,6 +29,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -99,7 +100,7 @@ public class TimerMessageStoreTest {
         storeConfig.setTimerInterceptDelayLevel(true);
         storeConfig.setTimerPrecisionMs(precisionMs);
 
-        messageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("TimerTest",false), new MyMessageArrivingListener(), new BrokerConfig());
+        messageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("TimerTest",false), new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
         boolean load = messageStore.load();
         assertTrue(load);
         messageStore.start();