[ISSUE #9777 ]feat: use data version from master while sync slave and fix delete co… (#9778)

absolute8511 · web-flow · commit 2b5e7094d256 · 2026-04-09T16:38:47.000+08:00
* feat: use data version from master while sync slave and fix delete config while sync

Change-Id: I42b2e7b1acc6836d3c90973801c9defba5f1325c

* fix: assign new version using master while sync slave

Change-Id: I7ec20607a84499fe5a6607763013c59d726aedc3

* feat: allow set dataVersion directly for topic/group config sync

Change-Id: Ic845794350e8bdaa847bdd0ae4b3e40ab1ad6311

* feat: set data version directly while sync from master

Change-Id: I39e78477a5223b578a4ede3e5cb76f04368d1ca3

* test: adjust slave sync test for version

Change-Id: I9e835568912928ddf6e81816095ee3ed8f93afc0
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConfigManager.java
@@ -165,6 +165,12 @@ public void updateKvDataVersion() throws Exception {
             JSON.toJSONString(kvDataVersion).getBytes(StandardCharsets.UTF_8));
     }
 
+    public void setKvDataVersion(DataVersion dataVersion)  throws Exception {
+        this.kvDataVersion = dataVersion;
+        this.configRocksDBStorage.put(versionCF, KV_DATA_VERSION_KEY, KV_DATA_VERSION_KEY.length,
+            JSON.toJSONString(kvDataVersion).getBytes(StandardCharsets.UTF_8));
+    }
+    
     public DataVersion getKvDataVersion() {
         return kvDataVersion;
     }
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBSubscriptionGroupManager.java b/broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBSubscriptionGroupManager.java
@@ -250,6 +250,16 @@ public void updateDataVersion() {
         }
     }
 
+    @Override
+    public void setDataVersion(DataVersion dataVersion) {
+        try {
+            rocksDBConfigManager.setKvDataVersion(dataVersion);
+        } catch (Exception e) {
+            log.error("set group config dataVersion error", e);
+            throw new RuntimeException(e);
+        }
+    }
+
     protected void decodeForbidden(byte[] key, byte[] body) {
         String forbiddenGroupName = new String(key, RocksDBConfigManager.CHARSET);
         JSONObject jsonObject = JSON.parseObject(new String(body, RocksDBConfigManager.CHARSET));
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBTopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBTopicConfigManager.java
@@ -205,6 +205,16 @@ public void updateDataVersion() {
         }
     }
 
+    @Override
+    public void setDataVersion(DataVersion dataVersion) {
+        try {
+            rocksDBConfigManager.setKvDataVersion(dataVersion);
+        } catch (Exception e) {
+            log.error("set topic config dataVersion error", e);
+            throw new RuntimeException(e);
+        }
+    }
+    
     /**
      * Migrate data from separate RocksDB instances to the unified RocksDB when useSingleRocksDBForAllConfigs is
      * enabled.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java b/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
@@ -83,8 +83,6 @@ private void syncTopicConfig() {
                 TopicConfigManager topicConfigManager = this.brokerController.getTopicConfigManager();
                 if (!topicConfigManager.getDataVersion().equals(topicWrapper.getDataVersion())) {
 
-                    topicConfigManager.getDataVersion().assignNewOne(topicWrapper.getDataVersion());
-
                     ConcurrentMap<String, TopicConfig> newTopicConfigTable = topicWrapper.getTopicConfigTable();
                     ConcurrentMap<String, TopicConfig> topicConfigTable = topicConfigManager.getTopicConfigTable();
 
@@ -94,13 +92,13 @@ private void syncTopicConfig() {
                         Map.Entry<String, TopicConfig> entry = iterator.next();
                         if (!newTopicConfigTable.containsKey(entry.getKey())) {
                             iterator.remove();
+                            topicConfigManager.deleteTopicConfig(entry.getKey());
                         }
-                        topicConfigManager.deleteTopicConfig(entry.getKey());
                     }
 
                     //update
                     newTopicConfigTable.values().forEach(topicConfigManager::putTopicConfig);
-                    topicConfigManager.updateDataVersion();
+                    topicConfigManager.setDataVersion(topicWrapper.getDataVersion());
                     topicConfigManager.persist();
                 }
                 if (topicWrapper.getTopicQueueMappingDetailMap() != null
@@ -177,7 +175,6 @@ private void syncSubscriptionGroupConfig() {
                 if (!this.brokerController.getSubscriptionGroupManager().getDataVersion()
                         .equals(subscriptionWrapper.getDataVersion())) {
                     SubscriptionGroupManager subscriptionGroupManager = this.brokerController.getSubscriptionGroupManager();
-                    subscriptionGroupManager.getDataVersion().assignNewOne(subscriptionWrapper.getDataVersion());
 
                     ConcurrentMap<String, SubscriptionGroupConfig> curSubscriptionGroupTable =
                             subscriptionGroupManager.getSubscriptionGroupTable();
@@ -189,12 +186,12 @@ private void syncSubscriptionGroupConfig() {
                         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);
-                    subscriptionGroupManager.updateDataVersion();
+                    subscriptionGroupManager.setDataVersion(subscriptionWrapper.getDataVersion());
                     // persist
                     subscriptionGroupManager.persist();
                     LOGGER.info("Update slave Subscription Group from master, {}", masterAddrBak);
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
@@ -778,6 +778,8 @@ public void updateDataVersion() {
         dataVersion.nextVersion(stateMachineVersion);
     }
 
-
+    public void setDataVersion(DataVersion dataVersion) {
+        this.dataVersion.assignNewOne(dataVersion);
+    }
 
 }
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java b/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java
@@ -55,6 +55,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -113,13 +115,9 @@ public void init() {
         when(brokerController.getMessageStore()).thenReturn(messageStore);
         when(brokerController.getTimerMessageStore()).thenReturn(timerMessageStore);
         when(brokerController.getTimerCheckpoint()).thenReturn(timerCheckpoint);
-        when(topicConfigManager.getDataVersion()).thenReturn(new DataVersion());
-        when(topicConfigManager.getTopicConfigTable()).thenReturn(new ConcurrentHashMap<>());
         when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
         when(consumerOffsetManager.getOffsetTable()).thenReturn(new ConcurrentHashMap<>());
         when(consumerOffsetManager.getDataVersion()).thenReturn(new DataVersion());
-        when(subscriptionGroupManager.getDataVersion()).thenReturn(new DataVersion());
-        when(subscriptionGroupManager.getSubscriptionGroupTable()).thenReturn(new ConcurrentHashMap<>());
         when(queryAssignmentProcessor.getMessageRequestModeManager()).thenReturn(messageRequestModeManager);
         when(messageRequestModeManager.getMessageRequestModeMap()).thenReturn(new ConcurrentHashMap<>());
         when(messageStoreConfig.isTimerWheelEnable()).thenReturn(true);
@@ -136,17 +134,31 @@ public void init() {
     public void testSyncAll() throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
         MQBrokerException, InterruptedException, UnsupportedEncodingException, RemotingCommandException {
         TopicConfig newTopicConfig = new TopicConfig("NewTopic");
-        when(brokerOuterAPI.getAllTopicConfig(anyString())).thenReturn(createTopicConfigWrapper(newTopicConfig));
+        TopicConfigAndMappingSerializeWrapper topicConfigWrapper = createTopicConfigWrapper(newTopicConfig);
+        when(brokerOuterAPI.getAllTopicConfig(anyString())).thenReturn(topicConfigWrapper);
         when(brokerOuterAPI.getAllConsumerOffset(anyString())).thenReturn(createConsumerOffsetWrapper());
         when(brokerOuterAPI.getAllDelayOffset(anyString())).thenReturn("");
-        when(brokerOuterAPI.getAllSubscriptionGroupConfig(anyString())).thenReturn(createSubscriptionGroupWrapper());
+        SubscriptionGroupWrapper subscriptionGroupWrapper = createSubscriptionGroupWrapper();
+        when(brokerOuterAPI.getAllSubscriptionGroupConfig(anyString())).thenReturn(subscriptionGroupWrapper);
         when(brokerOuterAPI.getAllMessageRequestMode(anyString())).thenReturn(createMessageRequestModeWrapper());
         when(brokerOuterAPI.getTimerMetrics(anyString())).thenReturn(createTimerMetricsWrapper());
+
+        TopicConfigManager topicConfigManager = new TopicConfigManager();
+        TopicConfigManager spiedTopicConfigManager = spy(topicConfigManager);
+        doNothing().when(spiedTopicConfigManager).persist();
+        SubscriptionGroupManager groupConfigManager = new SubscriptionGroupManager();
+        SubscriptionGroupManager spiedGroupConfigManager = spy(groupConfigManager);
+        doNothing().when(spiedGroupConfigManager).persist();
+        when(brokerController.getTopicConfigManager()).thenReturn(spiedTopicConfigManager);
+        when(brokerController.getSubscriptionGroupManager()).thenReturn(spiedGroupConfigManager);
+
         slaveSynchronize.syncAll();
-        Assert.assertEquals(1, this.brokerController.getTopicConfigManager().getDataVersion().getStateVersion());
-        Assert.assertEquals(1, this.brokerController.getTopicQueueMappingManager().getDataVersion().getStateVersion());
+        long topicVer = topicConfigWrapper.getDataVersion().getStateVersion();
+        long groupVer = subscriptionGroupWrapper.getDataVersion().getStateVersion();
+        Assert.assertEquals(topicVer, this.brokerController.getTopicConfigManager().getDataVersion().getStateVersion());
+        Assert.assertEquals(topicVer, this.brokerController.getTopicQueueMappingManager().getDataVersion().getStateVersion());
         Assert.assertEquals(1, consumerOffsetManager.getDataVersion().getStateVersion());
-        Assert.assertEquals(1, subscriptionGroupManager.getDataVersion().getStateVersion());
+        Assert.assertEquals(groupVer, this.brokerController.getSubscriptionGroupManager().getDataVersion().getStateVersion());
         Assert.assertEquals(1, timerMetrics.getDataVersion().getStateVersion());
     }
 
@@ -167,7 +179,7 @@ private TopicConfigAndMappingSerializeWrapper createTopicConfigWrapper(TopicConf
         wrapper.setTopicConfigTable(new ConcurrentHashMap<>());
         wrapper.getTopicConfigTable().put(topicConfig.getTopicName(), topicConfig);
         DataVersion dataVersion = new DataVersion();
-        dataVersion.setStateVersion(1L);
+        dataVersion.setStateVersion(5L);
         wrapper.setDataVersion(dataVersion);
         wrapper.setMappingDataVersion(dataVersion);
         return wrapper;
@@ -186,7 +198,7 @@ private SubscriptionGroupWrapper createSubscriptionGroupWrapper() {
         SubscriptionGroupWrapper wrapper = new SubscriptionGroupWrapper();
         wrapper.setSubscriptionGroupTable(new ConcurrentHashMap<>());
         DataVersion dataVersion = new DataVersion();
-        dataVersion.setStateVersion(1L);
+        dataVersion.setStateVersion(5L);
         wrapper.setDataVersion(dataVersion);
         return wrapper;
     }

Original file line number	Diff line number	Diff line change
`@@ -165,6 +165,12 @@ public void updateKvDataVersion() throws Exception {`
`165`	`165`	`JSON.toJSONString(kvDataVersion).getBytes(StandardCharsets.UTF_8));`
`166`	`166`	`}`
`167`	`167`
	`168`	`+ public void setKvDataVersion(DataVersion dataVersion) throws Exception {`
	`169`	`+ this.kvDataVersion = dataVersion;`
	`170`	`+ this.configRocksDBStorage.put(versionCF, KV_DATA_VERSION_KEY, KV_DATA_VERSION_KEY.length,`
	`171`	`+ JSON.toJSONString(kvDataVersion).getBytes(StandardCharsets.UTF_8));`
	`172`	`+ }`
	`173`	`+`
`168`	`174`	`public DataVersion getKvDataVersion() {`
`169`	`175`	`return kvDataVersion;`
`170`	`176`	`}`
Original file line number	Diff line number	Diff line change
`@@ -778,6 +778,8 @@ public void updateDataVersion() {`
`778`	`778`	`dataVersion.nextVersion(stateMachineVersion);`
`779`	`779`	`}`
`780`	`780`
`781`		`-`
	`781`	`+ public void setDataVersion(DataVersion dataVersion) {`
	`782`	`+ this.dataVersion.assignNewOne(dataVersion);`
	`783`	`+ }`
`782`	`784`
`783`	`785`	`}`