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

@@ -93,9 +93,8 @@ private void syncTopicConfig() {
                     while (iterator.hasNext()) {
                         Map.Entry<String, TopicConfig> entry = iterator.next();
                         if (!newTopicConfigTable.containsKey(entry.getKey())) {
-                            iterator.remove();
+                            topicConfigManager.deleteTopicConfig(entry.getKey());
                         }
-                        topicConfigManager.deleteTopicConfig(entry.getKey());
                     }
 
                     //update
@@ -188,9 +187,8 @@ private void syncSubscriptionGroupConfig() {
                     while (iterator.hasNext()) {
                         Map.Entry<String, SubscriptionGroupConfig> configEntry = iterator.next();
                         if (!newSubscriptionGroupTable.containsKey(configEntry.getKey())) {
-                            iterator.remove();
+                            subscriptionGroupManager.deleteSubscriptionGroupConfig(configEntry.getKey());
                         }
-                        subscriptionGroupManager.deleteSubscriptionGroupConfig(configEntry.getKey());
                     }
                     // update
                     newSubscriptionGroupTable.values().forEach(subscriptionGroupManager::putSubscriptionGroupConfig);

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

@@ -162,6 +162,50 @@ 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());
+
+        Assert.assertEquals(0, topicConfigManager.getDataVersion().getStateVersion());
+        slaveSynchronize.syncAll();
+
+        Assert.assertEquals(1, topicConfigManager.getDataVersion().getStateVersion());
+        Assert.assertTrue(topicConfigManager.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);
+        when(topicConfigManager.getTopicConfigTable()).thenReturn(localTable);
+        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(topicConfigManager.getTopicConfigTable().containsKey("LocalTopic"));
+        Assert.assertTrue(topicConfigManager.getTopicConfigTable().containsKey("NewTopic"));
+    }
+
     private TopicConfigAndMappingSerializeWrapper createTopicConfigWrapper(TopicConfig topicConfig) {
         TopicConfigAndMappingSerializeWrapper wrapper = new TopicConfigAndMappingSerializeWrapper();
         wrapper.setTopicConfigTable(new ConcurrentHashMap<>());