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

[Issue #4330]Fix some comments in DefaultElectPolicy

Browse Source
This commit is contained in:
hzh0425
2022-11-19 21:28:28 +08:00
committed by GitHub
parent 0b24e875b7
commit 44072e62bd
1 changed files with 18 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/src/main/java/org/apache/rocketmq/controller/elect/impl/DefaultElectPolicy.java
+18 -13
View File
@@ -29,10 +29,11 @@ import java.util.stream.Collectors;
public class DefaultElectPolicy implements ElectPolicy {
// <clusterName, brokerAddr> valid predicate
// <clusterName, brokerAddr>, Used to judge whether a broker
// has preliminary qualification to be selected as master
private BiPredicate<String, String> validPredicate;
// <clusterName, brokerAddr, info> getter to get more information
// <clusterName, brokerAddr, BrokerLiveInfo>, Used to obtain the BrokerLiveInfo information of a broker
private BiFunction<String, String, BrokerLiveInfo> additionalInfoGetter;
private final Comparator<BrokerLiveInfo> comparator = (x, y) -> {
@@ -49,10 +50,13 @@ public class DefaultElectPolicy implements ElectPolicy {
}
/**
* try to elect a master, if old master still alive, now we do nothing,
* if preferBrokerAddr is not blank, that means we must elect a new master,
* and we should check if the preferBrokerAddr is valid, if so we should elect it as
* new master, if else we should elect nothing.
* We will try to select a new master from syncStateBrokers and allReplicaBrokers in turn.
* The strategies are as follows:
* - Filter alive brokers by 'validPredicate'.
* - Check whether the old master is still valid.
* - If preferBrokerAddr is not empty and valid, select it as master.
* - Otherwise, we will sort the array of 'brokerLiveInfo' according to (epoch, offset), and select the best candidate as the new master.
*
* @param clusterName the brokerGroup belongs
* @param syncStateBrokers all broker replicas in syncStateSet
* @param allReplicaBrokers all broker replicas
@@ -70,7 +74,8 @@ public class DefaultElectPolicy implements ElectPolicy {
if (StringUtils.isNotEmpty(newMaster)) {
return newMaster;
}
// try to elect in all replicas
// try to elect in all allReplicaBrokers
if (allReplicaBrokers != null) {
newMaster = tryElect(clusterName, allReplicaBrokers, oldMaster, preferBrokerAddr);
}
@@ -82,19 +87,19 @@ public class DefaultElectPolicy implements ElectPolicy {
if (this.validPredicate != null) {
brokers = brokers.stream().filter(brokerAddr -> this.validPredicate.test(clusterName, brokerAddr)).collect(Collectors.toSet());
}
// try to elect in brokers
if (brokers.size() >= 1) {
if (!brokers.isEmpty()) {
// if old master is still valid, and preferBrokerAddr is blank or is equals to oldMaster
if (brokers.contains(oldMaster) && (StringUtils.isBlank(preferBrokerAddr) || preferBrokerAddr.equals(oldMaster))) {
// old master still valid, and our preferBrokerAddr is blank or is equals to oldMaster
return oldMaster;
}
// if preferBrokerAddr is not blank, if preferBrokerAddr is valid, we choose it, else we choose nothing
// if preferBrokerAddr is valid, we choose it, otherwise we choose nothing
if (StringUtils.isNotBlank(preferBrokerAddr)) {
return brokers.contains(preferBrokerAddr) ? preferBrokerAddr : null;
}
if (this.additionalInfoGetter != null) {
// get more information from getter
// sort brokerLiveInfos by epoch, maxOffset
// sort brokerLiveInfos by (epoch,maxOffset)
TreeSet<BrokerLiveInfo> brokerLiveInfos = new TreeSet<>(this.comparator);
brokers.forEach(brokerAddr -> brokerLiveInfos.add(this.additionalInfoGetter.apply(clusterName, brokerAddr)));
if (brokerLiveInfos.size() >= 1) {