diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java index f251e2b009..8347f36534 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java @@ -109,7 +109,7 @@ import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache; import org.apache.rocketmq.remoting.protocol.body.GetConsumerStatusBody; import org.apache.rocketmq.remoting.protocol.body.GroupList; import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo; -import org.apache.rocketmq.remoting.protocol.body.InSyncStateData; +import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo; import org.apache.rocketmq.remoting.protocol.body.KVTable; import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody; import org.apache.rocketmq.remoting.protocol.body.LockBatchResponseBody; @@ -2982,7 +2982,7 @@ public class MQClientAPIImpl implements NameServerUpdateCallback { throw new MQBrokerException(response.getCode(), response.getRemark()); } - public InSyncStateData getInSyncStateData(final String controllerAddress, + public BrokerReplicasInfo getInSyncStateData(final String controllerAddress, final List brokers) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException, RemotingCommandException { // Get controller leader address. final GetMetaDataResponseHeader controllerMetaData = getControllerMetaData(controllerAddress); @@ -2997,7 +2997,7 @@ public class MQClientAPIImpl implements NameServerUpdateCallback { assert response != null; switch (response.getCode()) { case ResponseCode.SUCCESS: { - return RemotingSerializable.decode(response.getBody(), InSyncStateData.class); + return RemotingSerializable.decode(response.getBody(), BrokerReplicasInfo.class); } default: break; diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java index a820b069e6..1c5a805b62 100644 --- a/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java +++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java @@ -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 getSyncStateData(final List brokerNames) { final ControllerResult 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 syncStateSet = syncStateInfo.getSyncStateSet(); final String master = syncStateInfo.getMasterAddress(); - final ArrayList 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 inSyncReplicas = new ArrayList<>(); + final ArrayList 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; } diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/InSyncStateData.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/BrokerReplicasInfo.java similarity index 59% rename from remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/InSyncStateData.java rename to remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/BrokerReplicasInfo.java index 2496f260a6..fece50d2e6 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/InSyncStateData.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/BrokerReplicasInfo.java @@ -21,38 +21,41 @@ import java.util.List; import java.util.Map; import org.apache.rocketmq.remoting.protocol.RemotingSerializable; -public class InSyncStateData extends RemotingSerializable { - private Map inSyncStateTable; +public class BrokerReplicasInfo extends RemotingSerializable { + private Map replicasInfoTable; - public InSyncStateData() { - this.inSyncStateTable = new HashMap<>(); + public BrokerReplicasInfo() { + this.replicasInfoTable = new HashMap<>(); } - public void addInSyncState(final String brokerName, final InSyncStateSet inSyncState) { - this.inSyncStateTable.put(brokerName, inSyncState); + public void addReplicaInfo(final String brokerName, final ReplicasInfo replicasInfo) { + this.replicasInfoTable.put(brokerName, replicasInfo); } - public Map getInSyncStateTable() { - return inSyncStateTable; + public Map getReplicasInfoTable() { + return replicasInfoTable; } - public void setInSyncStateTable( - Map inSyncStateTable) { - this.inSyncStateTable = inSyncStateTable; + public void setReplicasInfoTable( + Map replicasInfoTable) { + this.replicasInfoTable = replicasInfoTable; } - public static class InSyncStateSet extends RemotingSerializable { + public static class ReplicasInfo extends RemotingSerializable { private String masterAddress; private int masterEpoch; private int syncStateSetEpoch; - private List inSyncMembers; + private List inSyncReplicas; + private List notInSyncReplicas; - public InSyncStateSet(String masterAddress, int masterEpoch, int syncStateSetEpoch, - List inSyncMembers) { + public ReplicasInfo(String masterAddress, int masterEpoch, int syncStateSetEpoch, + List inSyncReplicas, + List notInSyncReplicas) { this.masterAddress = masterAddress; this.masterEpoch = masterEpoch; this.syncStateSetEpoch = syncStateSetEpoch; - this.inSyncMembers = inSyncMembers; + this.inSyncReplicas = inSyncReplicas; + this.notInSyncReplicas = notInSyncReplicas; } public String getMasterAddress() { @@ -79,21 +82,30 @@ public class InSyncStateData extends RemotingSerializable { this.syncStateSetEpoch = syncStateSetEpoch; } - public List getInSyncMembers() { - return inSyncMembers; + public List getInSyncReplicas() { + return inSyncReplicas; } - public void setInSyncMembers( - List inSyncMembers) { - this.inSyncMembers = inSyncMembers; + public void setInSyncReplicas( + List inSyncReplicas) { + this.inSyncReplicas = inSyncReplicas; + } + + public List getNotInSyncReplicas() { + return notInSyncReplicas; + } + + public void setNotInSyncReplicas( + List notInSyncReplicas) { + this.notInSyncReplicas = notInSyncReplicas; } } - public static class InSyncMember extends RemotingSerializable { + public static class ReplicaIdentity extends RemotingSerializable { private String address; private Long brokerId; - public InSyncMember(String address, Long brokerId) { + public ReplicaIdentity(String address, Long brokerId) { this.address = address; this.brokerId = brokerId; } @@ -116,7 +128,7 @@ public class InSyncStateData extends RemotingSerializable { @Override public String toString() { - return "InSyncMember{" + + return "{" + "address='" + address + '\'' + ", brokerId=" + brokerId + '}'; diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java index 7bc3080363..f70580dc65 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java @@ -51,7 +51,7 @@ import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache; import org.apache.rocketmq.remoting.protocol.body.GroupList; import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo; -import org.apache.rocketmq.remoting.protocol.body.InSyncStateData; +import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo; import org.apache.rocketmq.remoting.protocol.body.KVTable; import org.apache.rocketmq.remoting.protocol.body.ProducerConnection; import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo; @@ -776,7 +776,7 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { } @Override - public InSyncStateData getInSyncStateData(String controllerAddress, + public BrokerReplicasInfo getInSyncStateData(String controllerAddress, List brokers) throws RemotingException, InterruptedException, MQBrokerException { return this.defaultMQAdminExtImpl.getInSyncStateData(controllerAddress, brokers); } diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java index 0460ed95b9..fc3e079fe7 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java @@ -86,7 +86,7 @@ import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache; import org.apache.rocketmq.remoting.protocol.body.GroupList; import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo; -import org.apache.rocketmq.remoting.protocol.body.InSyncStateData; +import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo; import org.apache.rocketmq.remoting.protocol.body.KVTable; import org.apache.rocketmq.remoting.protocol.body.ProducerConnection; import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo; @@ -1819,7 +1819,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner { } @Override - public InSyncStateData getInSyncStateData(String controllerAddress, + public BrokerReplicasInfo getInSyncStateData(String controllerAddress, List brokers) throws RemotingException, InterruptedException, MQBrokerException { return this.mqClientInstance.getMQClientAPIImpl().getInSyncStateData(controllerAddress, brokers); } diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java index ebf878f324..2d19af5f2b 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java @@ -48,7 +48,7 @@ import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache; import org.apache.rocketmq.remoting.protocol.body.GroupList; import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo; -import org.apache.rocketmq.remoting.protocol.body.InSyncStateData; +import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo; import org.apache.rocketmq.remoting.protocol.body.KVTable; import org.apache.rocketmq.remoting.protocol.body.ProducerConnection; import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo; @@ -411,7 +411,7 @@ public interface MQAdminExt extends MQAdmin { HARuntimeInfo getBrokerHAStatus(String brokerAddr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException; - InSyncStateData getInSyncStateData(String controllerAddress, + BrokerReplicasInfo getInSyncStateData(String controllerAddress, List brokers) throws RemotingException, InterruptedException, MQBrokerException; EpochEntryCache getBrokerEpochCache( diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/ha/GetSyncStateSetSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/ha/GetSyncStateSetSubCommand.java index e9699e7131..252dd99fb0 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/command/ha/GetSyncStateSetSubCommand.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/ha/GetSyncStateSetSubCommand.java @@ -24,7 +24,7 @@ import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.Option; import org.apache.commons.cli.Options; import org.apache.rocketmq.remoting.RPCHook; -import org.apache.rocketmq.remoting.protocol.body.InSyncStateData; +import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo; import org.apache.rocketmq.srvutil.ServerUtil; import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; import org.apache.rocketmq.tools.command.CommandUtil; @@ -114,15 +114,20 @@ public class GetSyncStateSetSubCommand implements SubCommand { private void printData(String controllerAddress, List brokerNames, DefaultMQAdminExt defaultMQAdminExt) throws Exception { if (brokerNames.size() > 0) { - final InSyncStateData syncStateData = defaultMQAdminExt.getInSyncStateData(controllerAddress, brokerNames); - final Map syncTable = syncStateData.getInSyncStateTable(); - for (Map.Entry next : syncTable.entrySet()) { - final List syncMembers = next.getValue().getInSyncMembers(); - System.out.printf("\n#brokerName\t%s\n#MasterAddr\t%s\n#MasterEpoch\t%d\n#SyncStateSetEpoch\t%d\n#SyncStateSetMemberNums\t%d\n", + final BrokerReplicasInfo brokerReplicasInfo = defaultMQAdminExt.getInSyncStateData(controllerAddress, brokerNames); + final Map replicasInfoTable = brokerReplicasInfo.getReplicasInfoTable(); + for (Map.Entry next : replicasInfoTable.entrySet()) { + final List inSyncReplicas = next.getValue().getInSyncReplicas(); + final List notInSyncReplicas = next.getValue().getNotInSyncReplicas(); + System.out.printf("\n#brokerName\t%s\n#MasterAddr\t%s\n#MasterEpoch\t%d\n#SyncStateSetEpoch\t%d\n#SyncStateSetNums\t%d\n", next.getKey(), next.getValue().getMasterAddress(), next.getValue().getMasterEpoch(), next.getValue().getSyncStateSetEpoch(), - syncMembers.size()); - for (InSyncStateData.InSyncMember member : syncMembers) { - System.out.printf("\n member:\t%s\n", member.toString()); + inSyncReplicas.size()); + for (BrokerReplicasInfo.ReplicaIdentity member : inSyncReplicas) { + System.out.printf("\n InSyncReplica:\t%s\n", member.toString()); + } + + for (BrokerReplicasInfo.ReplicaIdentity member : notInSyncReplicas) { + System.out.printf("\n NotInSyncReplica:\t%s\n", member.toString()); } } }