fix: slave delete all topic config and subGroup config when sync config from master

Author: humkum

broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java

@@ -17,7 +17,8 @@
 package org.apache.rocketmq.broker.slave;
 
 import java.io.IOException;
-import java.util.Iterator;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -87,14 +88,13 @@ private void syncTopicConfig() {
                     ConcurrentMap<String, TopicConfig> topicConfigTable = topicConfigManager.getTopicConfigTable();
 
                     //delete
-                    Iterator<Map.Entry<String, TopicConfig>> iterator = topicConfigTable.entrySet().iterator();
-                    while (iterator.hasNext()) {
-                        Map.Entry<String, TopicConfig> entry = iterator.next();
+                    List<String> toRemoveTopic = new ArrayList<>();
+                    for (Map.Entry<String, TopicConfig> entry : topicConfigTable.entrySet()) {
                         if (!newTopicConfigTable.containsKey(entry.getKey())) {
-                            iterator.remove();
-                            topicConfigManager.deleteTopicConfig(entry.getKey());
+                            toRemoveTopic.add(entry.getKey());
                         }
                     }
+                    toRemoveTopic.forEach(topicConfigManager::deleteTopicConfig);
 
                     //update
                     newTopicConfigTable.values().forEach(topicConfigManager::putTopicConfig);
@@ -181,14 +181,14 @@ private void syncSubscriptionGroupConfig() {
                     ConcurrentMap<String, SubscriptionGroupConfig> newSubscriptionGroupTable =
                             subscriptionWrapper.getSubscriptionGroupTable();
                     // delete
-                    Iterator<Map.Entry<String, SubscriptionGroupConfig>> iterator = curSubscriptionGroupTable.entrySet().iterator();
-                    while (iterator.hasNext()) {
-                        Map.Entry<String, SubscriptionGroupConfig> configEntry = iterator.next();
+                    List<String> toRemoveSubscriptionGroup = new ArrayList<>();
+                    for (Map.Entry<String, SubscriptionGroupConfig> configEntry : curSubscriptionGroupTable.entrySet()) {
                         if (!newSubscriptionGroupTable.containsKey(configEntry.getKey())) {
-                            iterator.remove();
-                            subscriptionGroupManager.deleteSubscriptionGroupConfig(configEntry.getKey());
+                            toRemoveSubscriptionGroup.add(configEntry.getKey());
                         }
                     }
+                    toRemoveSubscriptionGroup.forEach(subscriptionGroupManager::deleteSubscriptionGroupConfig);
+
                     // update
                     newSubscriptionGroupTable.values().forEach(subscriptionGroupManager::putSubscriptionGroupConfig);
                     subscriptionGroupManager.setDataVersion(subscriptionWrapper.getDataVersion());

broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java

@@ -143,10 +143,10 @@ public void testSyncAll() throws RemotingConnectException, RemotingSendRequestEx
         when(brokerOuterAPI.getAllMessageRequestMode(anyString())).thenReturn(createMessageRequestModeWrapper());
         when(brokerOuterAPI.getTimerMetrics(anyString())).thenReturn(createTimerMetricsWrapper());
 
-        TopicConfigManager topicConfigManager = new TopicConfigManager();
+        TopicConfigManager topicConfigManager = new TopicConfigManager(brokerController);
         TopicConfigManager spiedTopicConfigManager = spy(topicConfigManager);
         doNothing().when(spiedTopicConfigManager).persist();
-        SubscriptionGroupManager groupConfigManager = new SubscriptionGroupManager();
+        SubscriptionGroupManager groupConfigManager = new SubscriptionGroupManager(brokerController);
         SubscriptionGroupManager spiedGroupConfigManager = spy(groupConfigManager);
         doNothing().when(spiedGroupConfigManager).persist();
         when(brokerController.getTopicConfigManager()).thenReturn(spiedTopicConfigManager);
@@ -174,6 +174,70 @@ public void testSyncTimerCheckPoint() throws RemotingConnectException, RemotingS
         Assert.assertEquals(0, timerCheckpoint.getDataVersion().getStateVersion());
     }
 
+    @Test
+    public void testSyncAllIncludesTopicConfig() throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
+        MQBrokerException, InterruptedException, RemotingCommandException, UnsupportedEncodingException {
+        TopicConfig newTopicConfig = new TopicConfig("TestTopic");
+        TopicConfigAndMappingSerializeWrapper topicWrapper = createTopicConfigWrapper(newTopicConfig);
+
+        when(brokerOuterAPI.getAllTopicConfig(anyString())).thenReturn(topicWrapper);
+        when(brokerOuterAPI.getAllConsumerOffset(anyString())).thenReturn(createConsumerOffsetWrapper());
+        when(brokerOuterAPI.getAllDelayOffset(anyString())).thenReturn("");
+        when(brokerOuterAPI.getAllSubscriptionGroupConfig(anyString())).thenReturn(createSubscriptionGroupWrapper());
+        when(brokerOuterAPI.getAllMessageRequestMode(anyString())).thenReturn(createMessageRequestModeWrapper());
+        when(brokerOuterAPI.getTimerMetrics(anyString())).thenReturn(createTimerMetricsWrapper());
+
+        TopicConfigManager topicConfigManager = new TopicConfigManager(brokerController);
+        TopicConfigManager spiedTopicConfigManager = spy(topicConfigManager);
+        doNothing().when(spiedTopicConfigManager).persist();
+        when(brokerController.getTopicConfigManager()).thenReturn(spiedTopicConfigManager);
+        SubscriptionGroupManager groupConfigManager = new SubscriptionGroupManager(brokerController);
+        SubscriptionGroupManager spiedGroupConfigManager = spy(groupConfigManager);
+        doNothing().when(spiedGroupConfigManager).persist();
+        when(brokerController.getSubscriptionGroupManager()).thenReturn(spiedGroupConfigManager);
+
+        TopicConfigManager topicConfigManager1 = brokerController.getTopicConfigManager();
+        DataVersion dataVersion = topicConfigManager1.getDataVersion();
+        Assert.assertEquals(0, dataVersion.getStateVersion());
+        slaveSynchronize.syncAll();
+
+        Assert.assertEquals(5, brokerController.getTopicConfigManager().getDataVersion().getStateVersion());
+        Assert.assertTrue(brokerController.getTopicConfigManager().getTopicConfigTable().containsKey("TestTopic"));
+    }
+
+    @Test
+    public void testSyncTopicConfigWithTopicDeletion() throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
+        MQBrokerException, InterruptedException, RemotingCommandException, UnsupportedEncodingException {
+        TopicConfig localTopic = new TopicConfig("LocalTopic");
+        ConcurrentHashMap<String, TopicConfig> localTable = new ConcurrentHashMap<>();
+        localTable.put("LocalTopic", localTopic);
+
+        TopicConfigManager topicConfigManager = new TopicConfigManager(brokerController);
+        TopicConfigManager spiedTopicConfigManager = spy(topicConfigManager);
+        spiedTopicConfigManager.setTopicConfigTable(localTable);
+        doNothing().when(spiedTopicConfigManager).persist();
+        when(brokerController.getTopicConfigManager()).thenReturn(spiedTopicConfigManager);
+        SubscriptionGroupManager groupConfigManager = new SubscriptionGroupManager(brokerController);
+        SubscriptionGroupManager spiedGroupConfigManager = spy(groupConfigManager);
+        doNothing().when(spiedGroupConfigManager).persist();
+        when(brokerController.getSubscriptionGroupManager()).thenReturn(spiedGroupConfigManager);
+        when(brokerOuterAPI.getAllConsumerOffset(anyString())).thenReturn(createConsumerOffsetWrapper());
+        when(brokerOuterAPI.getAllDelayOffset(anyString())).thenReturn("");
+        when(brokerOuterAPI.getAllSubscriptionGroupConfig(anyString())).thenReturn(createSubscriptionGroupWrapper());
+        when(brokerOuterAPI.getAllMessageRequestMode(anyString())).thenReturn(createMessageRequestModeWrapper());
+        when(brokerOuterAPI.getTimerMetrics(anyString())).thenReturn(createTimerMetricsWrapper());
+
+        TopicConfig newTopicConfig = new TopicConfig("NewTopic");
+        TopicConfigAndMappingSerializeWrapper topicWrapper = createTopicConfigWrapper(newTopicConfig);
+
+        when(brokerOuterAPI.getAllTopicConfig(anyString())).thenReturn(topicWrapper);
+
+        slaveSynchronize.syncAll();
+
+        Assert.assertFalse(brokerController.getTopicConfigManager().getTopicConfigTable().containsKey("LocalTopic"));
+        Assert.assertTrue(brokerController.getTopicConfigManager().getTopicConfigTable().containsKey("NewTopic"));
+    }
+
     private TopicConfigAndMappingSerializeWrapper createTopicConfigWrapper(TopicConfig topicConfig) {
         TopicConfigAndMappingSerializeWrapper wrapper = new TopicConfigAndMappingSerializeWrapper();
         wrapper.setTopicConfigTable(new ConcurrentHashMap<>());