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

GetSyncStateSetSubCommand can also print that the broker is not in syncStateSet (#5935)

Browse Source
This commit is contained in:
rongtong
2023-01-28 10:08:51 +08:00
committed by GitHub
parent ae9f119431
commit 0ea930182c
7 changed files with 75 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java
+16 -10
View File
@@ -41,7 +41,7 @@ import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
import org.apache.rocketmq.remoting.protocol.body.InSyncStateData;
import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetResponseHeader;
@@ -264,7 +264,7 @@ public class ReplicasInfoManager {
canBeElectedAsMaster = syncStateInfo.getSyncStateSet().contains(brokerAddress) || this.controllerConfig.isEnableElectUncleanMaster();
}
if (!canBeElectedAsMaster) {
// still need to apply an ElectMasterEvent to tell the statemachine
// still need to apply an ElectMasterEvent to tell the statemachine
// that the master was shutdown and no new master was elected. set SyncStateInfo.masterAddress empty
final ElectMasterEvent event = new ElectMasterEvent(false, brokerName);
result.addEvent(event);
@@ -322,7 +322,7 @@ public class ReplicasInfoManager {
public ControllerResult<Void> getSyncStateData(final List<String> brokerNames) {
final ControllerResult<Void> result = new ControllerResult<>();
final InSyncStateData inSyncStateData = new InSyncStateData();
final BrokerReplicasInfo brokerReplicasInfo = new BrokerReplicasInfo();
for (String brokerName : brokerNames) {
if (isContainsBroker(brokerName)) {
// If exist broker metadata, just return metadata
@@ -330,17 +330,23 @@ public class ReplicasInfoManager {
final BrokerInfo brokerInfo = this.replicaInfoTable.get(brokerName);
final Set<String> syncStateSet = syncStateInfo.getSyncStateSet();
final String master = syncStateInfo.getMasterAddress();
final ArrayList<InSyncStateData.InSyncMember> inSyncMembers = new ArrayList<>();
syncStateSet.forEach(replicas -> {
long brokerId = StringUtils.equals(master, replicas) ? MixAll.MASTER_ID : brokerInfo.getBrokerId(replicas);
inSyncMembers.add(new InSyncStateData.InSyncMember(replicas, brokerId));
final ArrayList<BrokerReplicasInfo.ReplicaIdentity> inSyncReplicas = new ArrayList<>();
final ArrayList<BrokerReplicasInfo.ReplicaIdentity> notInSyncReplicas = new ArrayList<>();
brokerInfo.getBrokerIdTable().forEach((brokerAddress, brokerId) -> {
if (syncStateSet.contains(brokerAddress)) {
long id = StringUtils.equals(master, brokerAddress) ? MixAll.MASTER_ID : brokerInfo.getBrokerId(brokerAddress);
inSyncReplicas.add(new BrokerReplicasInfo.ReplicaIdentity(brokerAddress, id));
} else {
notInSyncReplicas.add(new BrokerReplicasInfo.ReplicaIdentity(brokerAddress, brokerId));
}
});
final InSyncStateData.InSyncStateSet inSyncState = new InSyncStateData.InSyncStateSet(master, syncStateInfo.getMasterEpoch(), syncStateInfo.getSyncStateSetEpoch(), inSyncMembers);
inSyncStateData.addInSyncState(brokerName, inSyncState);
final BrokerReplicasInfo.ReplicasInfo inSyncState = new BrokerReplicasInfo.ReplicasInfo(master, syncStateInfo.getMasterEpoch(), syncStateInfo.getSyncStateSetEpoch(), inSyncReplicas, notInSyncReplicas);
brokerReplicasInfo.addReplicaInfo(brokerName, inSyncState);
}
}
result.setBody(inSyncStateData.encode());
result.setBody(brokerReplicasInfo.encode());
return result;
}