opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-21 05:44:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #6316] Nameserver should choose a master with a larger epoch when there are two masters in controller mode (#6317)

Browse Source
This commit is contained in:
rongtong
2023-03-13 11:16:35 +08:00
committed by GitHub
parent d0df0518c7
commit 20dc5c96ea
7 changed files with 28 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
-2
View File
@@ -1548,8 +1548,6 @@ public class BrokerController {
this.brokerPreOnlineService.start();
}
//Init state version after messageStore initialized.
this.topicConfigManager.initStateVersion();
}
public void start() throws Exception {
broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java
+4
View File
@@ -207,6 +207,8 @@ public class ReplicasManager {
schedulingCheckSyncStateSet();
this.brokerController.getTopicConfigManager().getDataVersion().nextVersion(newMasterEpoch);
this.executorService.submit(() -> {
// Register broker to name-srv
try {
@@ -243,6 +245,8 @@ public class ReplicasManager {
// 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 {
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+2 -1
View File
@@ -777,7 +777,8 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
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 {
broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java
+2 -1
View File
@@ -195,7 +195,8 @@ public class SubscriptionGroupManager extends ConfigManager {
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();
}
broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
+8 -9
View File
@@ -294,7 +294,8 @@ public class TopicConfigManager extends ConfigManager {
}
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 class TopicConfigManager extends ConfigManager {
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 class TopicConfigManager extends ConfigManager {
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 class TopicConfigManager extends ConfigManager {
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 @@ public class TopicConfigManager extends ConfigManager {
public boolean containsTopic(String topic) {
return topicConfigTable.containsKey(topic);
}
}
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+9 -1
View File
@@ -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 class DefaultMessageStore implements MessageStore {
@Override
public long getStateMachineVersion() {
return 0L;
return stateMachineVersion;
}
public void setStateMachineVersion(long stateMachineVersion) {
this.stateMachineVersion = stateMachineVersion;
}
public BrokerStatsManager getBrokerStatsManager() {
@@ -3215,4 +3221,6 @@ public class DefaultMessageStore implements MessageStore {
return this.messageStoreConfig.isTransientStorePoolEnable() &&
(this.brokerConfig.isEnableControllerMode() || this.messageStoreConfig.getBrokerRole() != BrokerRole.SLAVE);
}
}
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
+3 -1
View File
@@ -147,8 +147,8 @@ public class AutoSwitchHAService extends DefaultHAService {
}
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 class AutoSwitchHAService extends DefaultHAService {
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) {