Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 1 commit into from
Mar 13, 2023
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
4 changes: 4 additions & 0 deletions broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
3 changes: 2 additions & 1 deletion broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
3 changes: 2 additions & 1 deletion broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand Down
17 changes: 8 additions & 9 deletions broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand Down Expand Up @@ -734,4 +731,6 @@ private String realKey(String key) {
public boolean containsTopic(String topic) {
return topicConfigTable.containsKey(topic);
}


}
10 changes: 9 additions & 1 deletion store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -3215,4 +3221,6 @@ public boolean isTransientStorePoolEnable() {
return this.messageStoreConfig.isTransientStorePoolEnable() &&
(this.brokerConfig.isEnableControllerMode() || this.messageStoreConfig.getBrokerRole() != BrokerRole.SLAVE);
}


}
4 changes: 3 additions & 1 deletion store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down Expand Up @@ -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) {
Expand Down