[ISSUE #10195] Ensure RocksDB compatibility in slave-master synchronization (#10196)

Author: gaoyf

broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java

@@ -323,7 +323,7 @@ public ConcurrentHashMap<String, SubscriptionGroupConfig> subGroupTable(String d
         int maxGroupNum) {
         // [groupSeq, groupSeq + maxGroupNum)
         int beginIndex = groupSeq;
-        if (StringUtils.isBlank(dataVersion) || !Objects.equals(DataVersion.fromJson(dataVersion, DataVersion.class), this.dataVersion)) {
+        if (beginIndex != 0 && (StringUtils.isBlank(dataVersion) || !Objects.equals(DataVersion.fromJson(dataVersion, DataVersion.class), getDataVersion()))) {
             beginIndex = 0;
             log.info("get sub subscription group table from {} due to {}", beginIndex,
                 StringUtils.isBlank(dataVersion) ? "DataVersion Empty" : "DataVersion Changed");

broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java

@@ -725,7 +725,7 @@ public ConcurrentHashMap<String, TopicConfig> subTopicConfigTable(String dataVer
         int maxTopicNum) {
         // [topicSeq, topicSeq + maxTopicNum)
         int beginIndex = topicSeq;
-        if (StringUtils.isBlank(dataVersion) || !Objects.equals(DataVersion.fromJson(dataVersion, DataVersion.class), this.dataVersion)) {
+        if (beginIndex != 0 && (StringUtils.isBlank(dataVersion) || !Objects.equals(DataVersion.fromJson(dataVersion, DataVersion.class), getDataVersion()))) {
             beginIndex = 0;
             log.info("get sub topic config table from {} due to {}", beginIndex,
                 StringUtils.isBlank(dataVersion) ? "DataVersion Empty" : "DataVersion Changed");