[ISSUE #6316] Nameserver should choose a master with a larger epoch w…

…hen there are two masters in controller mode (#6317)
apache · Mar 13, 2023 · 20dc5c9 · 20dc5c9
1 parent d0df051
commit 20dc5c9
Show file tree

Hide file tree

Showing 7 changed files with 28 additions and 15 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -1548,8 +1548,6 @@ protected void startBasicService() throws Exception {
             this.brokerPreOnlineService.start();
         }
 
-        //Init state version after messageStore initialized.
-        this.topicConfigManager.initStateVersion();
     }
 
     public void start() throws Exception {

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java b/broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java
@@ -207,6 +207,8 @@ public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch
 
                 schedulingCheckSyncStateSet();
 
+                this.brokerController.getTopicConfigManager().getDataVersion().nextVersion(newMasterEpoch);
+
                 this.executorService.submit(() -> {
                     // Register broker to name-srv
                     try {
@@ -243,6 +245,8 @@ public void changeToSlave(final String newMasterAddress, final int newMasterEpoc
                 // Notify ha service, change to slave
                 this.haService.changeToSlave(newMasterAddress, newMasterEpoch, this.brokerConfig.getBrokerId());
 
+                this.brokerController.getTopicConfigManager().getDataVersion().nextVersion(newMasterEpoch);
+
                 this.executorService.submit(() -> {
                     // Register broker to name-srv
                     try {

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -777,7 +777,8 @@ private synchronized RemotingCommand updateBrokerConfig(ChannelHandlerContext ct
                     LOGGER.info("updateBrokerConfig, new config: [{}] client: {} ", properties, ctx.channel().remoteAddress());
                     this.brokerController.getConfiguration().update(properties);
                     if (properties.containsKey("brokerPermission")) {
-                        this.brokerController.getTopicConfigManager().getDataVersion().nextVersion();
+                        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
+                        this.brokerController.getTopicConfigManager().getDataVersion().nextVersion(stateMachineVersion);
                         this.brokerController.registerBrokerAll(false, false, true);
                     }
                 } else {

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java b/broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java
@@ -195,7 +195,8 @@ private void updateForbiddenValue(String group, String topic, Integer forbidden)
             log.info("set group forbidden, {}@{} old: {} new: {}", group, topic, 0, forbidden);
         }
 
-        this.dataVersion.nextVersion();
+        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
+        dataVersion.nextVersion(stateMachineVersion);
 
         this.persist();
     }

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
@@ -294,7 +294,8 @@ public TopicConfig createTopicIfAbsent(TopicConfig topicConfig, boolean register
                     }
                     log.info("Create new topic [{}] config:[{}]", topicConfig.getTopicName(), topicConfig);
                     this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
-                    this.dataVersion.nextVersion();
+                    long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
+                    dataVersion.nextVersion(stateMachineVersion);
                     createNew = true;
                     this.persist();
                 } finally {
@@ -394,7 +395,8 @@ public TopicConfig createTopicOfTranCheckMaxTime(final int clientDefaultTopicQue
                     log.info("create new topic {}", topicConfig);
                     this.topicConfigTable.put(TopicValidator.RMQ_SYS_TRANS_CHECK_MAX_TIME_TOPIC, topicConfig);
                     createNew = true;
-                    this.dataVersion.nextVersion();
+                    long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
+                    dataVersion.nextVersion(stateMachineVersion);
                     this.persist();
                 } finally {
                     this.topicConfigTableLock.unlock();
@@ -540,7 +542,8 @@ public void deleteTopicConfig(final String topic) {
         TopicConfig old = this.topicConfigTable.remove(topic);
         if (old != null) {
             log.info("delete topic config OK, topic: {}", old);
-            this.dataVersion.nextVersion();
+            long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
+            dataVersion.nextVersion(stateMachineVersion);
             this.persist();
         } else {
             log.warn("delete topic config failed, topic: {} not exists", topic);
@@ -556,12 +559,6 @@ public TopicConfigSerializeWrapper buildTopicConfigSerializeWrapper() {
         return topicConfigSerializeWrapper;
     }
 
-    public void initStateVersion() {
-        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
-        dataVersion.nextVersion(stateMachineVersion);
-        this.persist();
-    }
-
     @Override
     public String encode() {
         return encode(false);
@@ -734,4 +731,6 @@ private String realKey(String key) {
     public boolean containsTopic(String topic) {
         return topicConfigTable.containsKey(topic);
     }
+
+
 }
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -198,6 +198,8 @@ public class DefaultMessageStore implements MessageStore {
 
     private final DispatchRequestOrderlyQueue dispatchRequestOrderlyQueue = new DispatchRequestOrderlyQueue(dispatchRequestOrderlyQueueSize);
 
+    private long stateMachineVersion = 0L;
+
     public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
         final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
         this.messageArrivingListener = messageArrivingListener;
@@ -1909,7 +1911,11 @@ public DispatchRequest checkMessageAndReturnSize(final ByteBuffer byteBuffer, fi
 
     @Override
     public long getStateMachineVersion() {
-        return 0L;
+        return stateMachineVersion;
+    }
+
+    public void setStateMachineVersion(long stateMachineVersion) {
+        this.stateMachineVersion = stateMachineVersion;
     }
 
     public BrokerStatsManager getBrokerStatsManager() {
@@ -3215,4 +3221,6 @@ public boolean isTransientStorePoolEnable() {
         return this.messageStoreConfig.isTransientStorePoolEnable() &&
             (this.brokerConfig.isEnableControllerMode() || this.messageStoreConfig.getBrokerRole() != BrokerRole.SLAVE);
     }
+
+
 }
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
@@ -147,8 +147,8 @@ public boolean changeToMaster(int masterEpoch) {
         }
 
         LOGGER.info("TruncateOffset is {}, confirmOffset is {}, maxPhyOffset is {}", truncateOffset, getConfirmOffset(), this.defaultMessageStore.getMaxPhyOffset());
-
         this.defaultMessageStore.recoverTopicQueueTable();
+        this.defaultMessageStore.setStateMachineVersion(masterEpoch);
         LOGGER.info("Change ha to master success, newMasterEpoch:{}, startOffset:{}", masterEpoch, newEpochEntry.getStartOffset());
         return true;
     }
@@ -178,6 +178,8 @@ public boolean changeToSlave(String newMasterAddr, int newMasterEpoch, Long slav
                 defaultMessageStore.getTransientStorePool().setRealCommit(false);
             }
 
+            this.defaultMessageStore.setStateMachineVersion(newMasterEpoch);
+
             LOGGER.info("Change ha to slave success, newMasterAddress:{}, newMasterEpoch:{}", newMasterAddr, newMasterEpoch);
             return true;
         } catch (final Exception e) {