You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2022/06/01 03:05:56 UTC
[rocketmq] branch 5.0.0-beta-dledger-controller updated: [Summer of code] Add admin tools for controller mode (#4388)
This is an automated email from the ASF dual-hosted git repository.
jinrongtong pushed a commit to branch 5.0.0-beta-dledger-controller
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/5.0.0-beta-dledger-controller by this push:
new e0121ca1d [Summer of code] Add admin tools for controller mode (#4388)
e0121ca1d is described below
commit e0121ca1da93f8ad77ff86db6b0dd261b32fd7fa
Author: hzh0425 <64...@qq.com>
AuthorDate: Wed Jun 1 11:05:48 2022 +0800
[Summer of code] Add admin tools for controller mode (#4388)
* add tool getSyncStateDataCommand
* get controller leaderAddr when execute command
* modify getControllerMetadata api
* code review
* code review
* add tool get brokerEpochCache
* init command
* set lastEpochEndOffset
* take maxPhyOffset in EpochCache
---
.../broker/hacontroller/ReplicasManager.java | 11 +-
.../broker/processor/AdminBrokerProcessor.java | 47 +++++---
.../rocketmq/client/impl/MQClientAPIImpl.java | 52 +++++++++
.../org/apache/rocketmq/common/EpochEntry.java | 4 +-
.../rocketmq/common/protocol/RequestCode.java | 4 +
.../common/protocol/body/EpochEntryCache.java | 87 ++++++++++++++
.../common/protocol/body/InSyncStateData.java | 124 ++++++++++++++++++++
.../org/apache/rocketmq/controller/Controller.java | 6 +
.../controller/impl/DledgerController.java | 7 ++
.../controller/impl/event/ControllerResult.java | 4 +
.../impl/manager/ReplicasInfoManager.java | 27 +++++
.../processor/ControllerRequestProcessor.java | 13 +++
.../apache/rocketmq/namesrv/NamesrvController.java | 7 +-
.../processor/ControllerRequestProcessor.java | 33 +++++-
.../store/ha/autoswitch/AutoSwitchHAService.java | 4 +
.../AutoSwitchRoleIntegrationTest.java | 1 +
.../rocketmq/tools/admin/DefaultMQAdminExt.java | 24 +++-
.../tools/admin/DefaultMQAdminExtImpl.java | 12 ++
.../apache/rocketmq/tools/admin/MQAdminExt.java | 17 ++-
.../apache/rocketmq/tools/command/CommandUtil.java | 9 ++
.../rocketmq/tools/command/MQAdminStartup.java | 10 +-
.../command/broker/GetBrokerEpochCacheCommand.java | 119 +++++++++++++++++++
.../tools/command/ha/SyncStateSetCommand.java | 130 +++++++++++++++++++++
23 files changed, 715 insertions(+), 37 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/hacontroller/ReplicasManager.java b/broker/src/main/java/org/apache/rocketmq/broker/hacontroller/ReplicasManager.java
index 41addf37c..fc0f252b4 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/hacontroller/ReplicasManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/hacontroller/ReplicasManager.java
@@ -31,6 +31,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.EpochEntry;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
@@ -333,9 +334,9 @@ public class ReplicasManager {
for (String address : this.controllerAddresses) {
try {
final GetMetaDataResponseHeader responseHeader = this.brokerOuterAPI.getControllerMetaData(address);
- if (responseHeader != null && responseHeader.isLeader()) {
- this.controllerLeaderAddress = address;
- LOGGER.info("Change controller leader address to {}", this.controllerAddresses);
+ if (responseHeader != null && StringUtils.isNoneEmpty(responseHeader.getControllerLeaderAddress())) {
+ this.controllerLeaderAddress = responseHeader.getControllerLeaderAddress();
+ LOGGER.info("Change controller leader address to {}", this.controllerLeaderAddress);
return true;
}
} catch (final Exception ignore) {
@@ -409,4 +410,8 @@ public class ReplicasManager {
public List<String> getControllerAddresses() {
return controllerAddresses;
}
+
+ public List<EpochEntry> getEpochEntries() {
+ return this.haService.getEpochEntries();
+ }
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index 0917858f7..3aaf4b935 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -20,7 +20,19 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
+import java.io.UnsupportedEncodingException;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
@@ -31,10 +43,12 @@ import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.filter.ConsumerFilterData;
import org.apache.rocketmq.broker.filter.ExpressionMessageFilter;
+import org.apache.rocketmq.broker.hacontroller.ReplicasManager;
import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.LockCallback;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
@@ -54,6 +68,7 @@ import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.RequestCode;
@@ -65,6 +80,7 @@ import org.apache.rocketmq.common.protocol.body.Connection;
import org.apache.rocketmq.common.protocol.body.ConsumeQueueData;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.common.protocol.body.EpochEntryCache;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.HARuntimeInfo;
import org.apache.rocketmq.common.protocol.body.KVTable;
@@ -152,7 +168,6 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
import org.apache.rocketmq.store.ConsumeQueueExt;
-import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageFilter;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
@@ -162,19 +177,6 @@ import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.ReferredIterator;
-import java.io.UnsupportedEncodingException;
-import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentMap;
-
import static org.apache.rocketmq.remoting.protocol.RemotingCommand.buildErrorResponse;
public class AdminBrokerProcessor implements NettyRequestProcessor {
@@ -295,6 +297,8 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
return this.getBrokerHaStatus(ctx, request);
case RequestCode.RESET_MASTER_FLUSH_OFFSET:
return this.resetMasterFlushOffset(ctx, request);
+ case RequestCode.GET_BROKER_EPOCH_CACHE:
+ return this.getBrokerEpochCache(ctx, request);
default:
return getUnknownCmdResponse(ctx, request);
}
@@ -2354,6 +2358,21 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
return response;
}
+ private RemotingCommand getBrokerEpochCache(ChannelHandlerContext ctx, RemotingCommand request) {
+ final ReplicasManager replicasManager = this.brokerController.getReplicasManager();
+ assert replicasManager != null;
+ final BrokerConfig brokerConfig = this.brokerController.getBrokerConfig();
+ final EpochEntryCache entryCache = new EpochEntryCache(brokerConfig.getBrokerClusterName(),
+ brokerConfig.getBrokerName(), brokerConfig.getBrokerId(), replicasManager.getEpochEntries(), this.brokerController.getMessageStore().getMaxPhyOffset());
+
+ final RemotingCommand response = RemotingCommand.createResponseCommand(null);
+ response.setBody(entryCache.encode());
+ response.setCode(ResponseCode.SUCCESS);
+ response.setRemark(null);
+ return response;
+ }
+
+
private RemotingCommand resetMasterFlushOffset(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
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 614ce9fb8..cf4fe19ba 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
@@ -85,9 +85,11 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.common.protocol.body.EpochEntryCache;
import org.apache.rocketmq.common.protocol.body.GetConsumerStatusBody;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.common.protocol.body.InSyncStateData;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.LockBatchRequestBody;
import org.apache.rocketmq.common.protocol.body.LockBatchResponseBody;
@@ -179,6 +181,7 @@ import org.apache.rocketmq.common.protocol.header.namesrv.GetRouteInfoRequestHea
import org.apache.rocketmq.common.protocol.header.namesrv.PutKVConfigRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.WipeWritePermOfBrokerRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.WipeWritePermOfBrokerResponseHeader;
+import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
@@ -207,6 +210,8 @@ import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import static org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode.SUCCESS;
+
public class MQClientAPIImpl implements NameServerUpdateCallback {
private final static InternalLogger log = ClientLogger.getLog();
private static boolean sendSmartMsg =
@@ -2854,4 +2859,51 @@ public class MQClientAPIImpl implements NameServerUpdateCallback {
throw new MQBrokerException(response.getCode(), response.getRemark());
}
+
+ public GetMetaDataResponseHeader getControllerMetaData(final String controllerAddress) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, RemotingCommandException, MQBrokerException {
+ final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_METADATA_INFO, null);
+ final RemotingCommand response = this.remotingClient.invokeSync(controllerAddress, request, 3000);
+ assert response != null;
+ if (response.getCode() == SUCCESS) {
+ return response.decodeCommandCustomHeader(GetMetaDataResponseHeader.class);
+ }
+ throw new MQBrokerException(response.getCode(), response.getRemark());
+ }
+
+ public InSyncStateData getInSyncStateData(final String controllerAddress,
+ final List<String> brokers) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException, RemotingCommandException {
+ // Get controller leader address.
+ final GetMetaDataResponseHeader controllerMetaData = getControllerMetaData(controllerAddress);
+ assert controllerMetaData != null;
+ assert controllerMetaData.getControllerLeaderAddress() != null;
+ final String leaderAddress = controllerMetaData.getControllerLeaderAddress();
+
+ RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_SYNC_STATE_DATA, null);
+ final byte[] body = RemotingSerializable.encode(brokers);
+ request.setBody(body);
+ RemotingCommand response = this.remotingClient.invokeSync(leaderAddress, request, 3000);
+ assert response != null;
+ switch (response.getCode()) {
+ case ResponseCode.SUCCESS: {
+ return RemotingSerializable.decode(response.getBody(), InSyncStateData.class);
+ }
+ default:
+ break;
+ }
+ throw new MQBrokerException(response.getCode(), response.getRemark());
+ }
+
+ public EpochEntryCache getBrokerEpochCache(String brokerAddr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
+ RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_EPOCH_CACHE, null);
+ final RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, 3000);
+ assert response != null;
+ switch (response.getCode()) {
+ case ResponseCode.SUCCESS: {
+ return RemotingSerializable.decode(response.getBody(), EpochEntryCache.class);
+ }
+ default:
+ break;
+ }
+ throw new MQBrokerException(response.getCode(), response.getRemark());
+ }
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/EpochEntry.java b/common/src/main/java/org/apache/rocketmq/common/EpochEntry.java
index 6cd3ceb23..8c71275b8 100644
--- a/common/src/main/java/org/apache/rocketmq/common/EpochEntry.java
+++ b/common/src/main/java/org/apache/rocketmq/common/EpochEntry.java
@@ -17,7 +17,9 @@
package org.apache.rocketmq.common;
-public class EpochEntry {
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class EpochEntry extends RemotingSerializable {
private int epoch;
private long startOffset;
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java b/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
index 7ba9fd046..b6313c583 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
@@ -250,4 +250,8 @@ public class RequestCode {
public static final int CONTROLLER_GET_REPLICA_INFO = 1004;
public static final int CONTROLLER_GET_METADATA_INFO = 1005;
+
+ public static final int CONTROLLER_GET_SYNC_STATE_DATA = 1006;
+
+ public static final int GET_BROKER_EPOCH_CACHE = 1007;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/body/EpochEntryCache.java b/common/src/main/java/org/apache/rocketmq/common/protocol/body/EpochEntryCache.java
new file mode 100644
index 000000000..cdfaa908d
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/body/EpochEntryCache.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.protocol.body;
+
+import java.util.List;
+import org.apache.rocketmq.common.EpochEntry;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class EpochEntryCache extends RemotingSerializable {
+ private String clusterName;
+ private String brokerName;
+ private long brokerId;
+ private List<EpochEntry> epochList;
+ private long maxOffset;
+
+ public EpochEntryCache(String clusterName, String brokerName, long brokerId, List<EpochEntry> epochList, long maxOffset) {
+ this.clusterName = clusterName;
+ this.brokerName = brokerName;
+ this.brokerId = brokerId;
+ this.epochList = epochList;
+ this.maxOffset = maxOffset;
+ }
+
+ public String getClusterName() {
+ return clusterName;
+ }
+
+ public void setClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ }
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+
+ public long getBrokerId() {
+ return brokerId;
+ }
+
+ public void setBrokerId(long brokerId) {
+ this.brokerId = brokerId;
+ }
+
+ public List<EpochEntry> getEpochList() {
+ return this.epochList;
+ }
+
+ public void setEpochList(List<EpochEntry> epochList) {
+ this.epochList = epochList;
+ }
+
+ public long getMaxOffset() {
+ return maxOffset;
+ }
+
+ public void setMaxOffset(long maxOffset) {
+ this.maxOffset = maxOffset;
+ }
+
+ @Override public String toString() {
+ return "EpochEntryCache{" +
+ "clusterName='" + clusterName + '\'' +
+ ", brokerName='" + brokerName + '\'' +
+ ", brokerId=" + brokerId +
+ ", epochList=" + epochList +
+ ", maxOffset=" + maxOffset +
+ '}';
+ }
+}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/body/InSyncStateData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/body/InSyncStateData.java
new file mode 100644
index 000000000..0d953c9d2
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/body/InSyncStateData.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.protocol.body;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class InSyncStateData extends RemotingSerializable {
+ private Map<String/*brokerName*/, InSyncState> inSyncTable;
+
+ public InSyncStateData() {
+ this.inSyncTable = new HashMap<>();
+ }
+
+ public void addInSyncState(final String brokerName, final InSyncState inSyncState) {
+ this.inSyncTable.put(brokerName, inSyncState);
+ }
+
+ public Map<String, InSyncState> getInSyncTable() {
+ return inSyncTable;
+ }
+
+ public void setInSyncTable(
+ Map<String, InSyncState> inSyncTable) {
+ this.inSyncTable = inSyncTable;
+ }
+
+ public static class InSyncState extends RemotingSerializable {
+ private String masterAddress;
+ private int masterEpoch;
+ private int syncStateSetEpoch;
+ private List<InSyncMember> inSyncMembers;
+
+ public InSyncState(String masterAddress, int masterEpoch, int syncStateSetEpoch,
+ List<InSyncMember> inSyncMembers) {
+ this.masterAddress = masterAddress;
+ this.masterEpoch = masterEpoch;
+ this.syncStateSetEpoch = syncStateSetEpoch;
+ this.inSyncMembers = inSyncMembers;
+ }
+
+ public String getMasterAddress() {
+ return masterAddress;
+ }
+
+ public void setMasterAddress(String masterAddress) {
+ this.masterAddress = masterAddress;
+ }
+
+ public int getMasterEpoch() {
+ return masterEpoch;
+ }
+
+ public void setMasterEpoch(int masterEpoch) {
+ this.masterEpoch = masterEpoch;
+ }
+
+ public int getSyncStateSetEpoch() {
+ return syncStateSetEpoch;
+ }
+
+ public void setSyncStateSetEpoch(int syncStateSetEpoch) {
+ this.syncStateSetEpoch = syncStateSetEpoch;
+ }
+
+ public List<InSyncMember> getInSyncMembers() {
+ return inSyncMembers;
+ }
+
+ public void setInSyncMembers(
+ List<InSyncMember> inSyncMembers) {
+ this.inSyncMembers = inSyncMembers;
+ }
+ }
+
+ public static class InSyncMember extends RemotingSerializable {
+ private String address;
+ private Long brokerId;
+
+ public InSyncMember(String address, Long brokerId) {
+ this.address = address;
+ this.brokerId = brokerId;
+ }
+
+ public String getAddress() {
+ return address;
+ }
+
+ public void setAddress(String address) {
+ this.address = address;
+ }
+
+ public Long getBrokerId() {
+ return brokerId;
+ }
+
+ public void setBrokerId(Long brokerId) {
+ this.brokerId = brokerId;
+ }
+
+ @Override public String toString() {
+ return "InSyncMember{" +
+ "address='" + address + '\'' +
+ ", brokerId=" + brokerId +
+ '}';
+ }
+ }
+}
diff --git a/controller/src/main/java/org/apache/rocketmq/controller/Controller.java b/controller/src/main/java/org/apache/rocketmq/controller/Controller.java
index fca8f5565..b4ed69a40 100644
--- a/controller/src/main/java/org/apache/rocketmq/controller/Controller.java
+++ b/controller/src/main/java/org/apache/rocketmq/controller/Controller.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.controller;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.common.protocol.body.SyncStateSet;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetRequestHeader;
@@ -88,6 +89,11 @@ public interface Controller {
*/
CompletableFuture<RemotingCommand> getReplicaInfo(final GetReplicaInfoRequestHeader request);
+ /**
+ * Get inSyncStateData for target brokers, this api is used for admin tools.
+ */
+ CompletableFuture<RemotingCommand> getSyncStateData(final List<String> brokerNames);
+
/**
* Get Metadata of controller
*
diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/DledgerController.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/DledgerController.java
index 8a4428f3b..eb85719d9 100644
--- a/controller/src/main/java/org/apache/rocketmq/controller/impl/DledgerController.java
+++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/DledgerController.java
@@ -154,6 +154,13 @@ public class DledgerController implements Controller {
() -> this.replicasInfoManager.getReplicaInfo(request), false);
}
+ @Override
+ public CompletableFuture<RemotingCommand> getSyncStateData(List<String> brokerNames) {
+
+ return this.scheduler.appendEvent("getSyncStateData",
+ () -> this.replicasInfoManager.getSyncStateData(brokerNames), false);
+ }
+
@Override
public RemotingCommand getControllerMetadata() {
final MemberState state = getMemberState();
diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/event/ControllerResult.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/event/ControllerResult.java
index 41af2af04..5c6318434 100644
--- a/controller/src/main/java/org/apache/rocketmq/controller/impl/event/ControllerResult.java
+++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/event/ControllerResult.java
@@ -27,6 +27,10 @@ public class ControllerResult<T> {
private int responseCode = ResponseCode.SUCCESS;
private String remark;
+ public ControllerResult() {
+ this(null);
+ }
+
public ControllerResult(T response) {
this.events = new ArrayList<>();
this.response = response;
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 3f991df4e..a34aedd13 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
@@ -16,8 +16,10 @@
*/
package org.apache.rocketmq.controller.impl.manager;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiPredicate;
@@ -27,6 +29,7 @@ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.namesrv.ControllerConfig;
import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.InSyncStateData;
import org.apache.rocketmq.common.protocol.body.SyncStateSet;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetResponseHeader;
@@ -277,6 +280,30 @@ public class ReplicasInfoManager {
return result;
}
+ public ControllerResult<Void> getSyncStateData(final List<String> brokerNames) {
+ final ControllerResult<Void> result = new ControllerResult<>();
+ final InSyncStateData inSyncStateData = new InSyncStateData();
+ for (String brokerName : brokerNames) {
+ if (isContainsBroker(brokerName)) {
+ // If exist broker metadata, just return metadata
+ final InSyncReplicasInfo replicasInfo = this.inSyncReplicasInfoTable.get(brokerName);
+ final BrokerInfo brokerInfo = this.replicaInfoTable.get(brokerName);
+ final Set<String> syncStateSet = replicasInfo.getSyncStateSet();
+ final String master = replicasInfo.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 InSyncStateData.InSyncState inSyncState = new InSyncStateData.InSyncState(master, replicasInfo.getMasterEpoch(), replicasInfo.getSyncStateSetEpoch(), inSyncMembers);
+ inSyncStateData.addInSyncState(brokerName, inSyncState);
+ }
+ }
+ result.setBody(inSyncStateData.encode());
+ return result;
+ }
+
/**
* Apply events to memory statemachine.
*
diff --git a/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java b/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java
index bf185c826..164e9e701 100644
--- a/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java
+++ b/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.controller.processor;
import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.constant.LoggerName;
@@ -43,6 +44,7 @@ import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_ALTER_S
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_ELECT_MASTER;
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_GET_METADATA_INFO;
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_GET_REPLICA_INFO;
+import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_GET_SYNC_STATE_DATA;
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_REGISTER_BROKER;
/**
@@ -114,6 +116,17 @@ public class ControllerRequestProcessor implements NettyRequestProcessor {
final BrokerHeartbeatRequestHeader requestHeader = request.decodeCommandCustomHeader(BrokerHeartbeatRequestHeader.class);
this.heartbeatManager.onBrokerHeartbeat(requestHeader.getClusterName(), requestHeader.getBrokerAddr());
}
+ case CONTROLLER_GET_SYNC_STATE_DATA: {
+ if (request.getBody() != null) {
+ final List<String> brokerNames = RemotingSerializable.decode(request.getBody(), List.class);
+ if (brokerNames != null && brokerNames.size() > 0) {
+ final CompletableFuture<RemotingCommand> future = this.controller.getSyncStateData(brokerNames);
+ if (future != null) {
+ return future.get(WAIT_TIMEOUT_OUT, TimeUnit.SECONDS);
+ }
+ }
+ }
+ }
default: {
final String error = " request type " + request.getCode() + " not supported";
return RemotingCommand.createResponseCommand(ResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvController.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvController.java
index 3f2e0f7f4..b6a109a61 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvController.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvController.java
@@ -270,12 +270,13 @@ public class NamesrvController {
this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.defaultExecutor);
if (controllerConfig.isEnableStartupController()) {
- final ControllerRequestProcessor controllerRequestProcessor = new ControllerRequestProcessor(this.controller);
+ final ControllerRequestProcessor controllerRequestProcessor = new ControllerRequestProcessor(this);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET, controllerRequestProcessor, this.controllerRequestExecutor);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_ELECT_MASTER, controllerRequestProcessor, this.controllerRequestExecutor);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_REGISTER_BROKER, controllerRequestProcessor, this.controllerRequestExecutor);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_GET_REPLICA_INFO, controllerRequestProcessor, this.controllerRequestExecutor);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_GET_METADATA_INFO, controllerRequestProcessor, this.controllerRequestExecutor);
+ this.remotingServer.registerProcessor(RequestCode.CONTROLLER_GET_SYNC_STATE_DATA, controllerRequestProcessor, this.controllerRequestExecutor);
}
}
}
@@ -344,6 +345,10 @@ public class NamesrvController {
this.remotingServer = remotingServer;
}
+ public Controller getController() {
+ return controller;
+ }
+
public Configuration getConfiguration() {
return configuration;
}
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ControllerRequestProcessor.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ControllerRequestProcessor.java
index 01e75803d..2651636e3 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ControllerRequestProcessor.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ControllerRequestProcessor.java
@@ -17,18 +17,22 @@
package org.apache.rocketmq.namesrv.processor;
import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.SyncStateSet;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.BrokerRegisterRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterRequestHeader;
+import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.controller.Controller;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
@@ -38,6 +42,7 @@ import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_ALTER_S
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_ELECT_MASTER;
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_GET_METADATA_INFO;
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_GET_REPLICA_INFO;
+import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_GET_SYNC_STATE_DATA;
import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_REGISTER_BROKER;
/**
@@ -46,10 +51,12 @@ import static org.apache.rocketmq.common.protocol.RequestCode.CONTROLLER_REGISTE
public class ControllerRequestProcessor implements NettyRequestProcessor {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private static final int WAIT_TIMEOUT_OUT = 5;
+ private final NamesrvController namesrvController;
private final Controller controller;
- public ControllerRequestProcessor(final Controller controller) {
- this.controller = controller;
+ public ControllerRequestProcessor(final NamesrvController namesrvController) {
+ this.namesrvController = namesrvController;
+ this.controller = namesrvController.getController();
}
@Override
@@ -95,7 +102,27 @@ public class ControllerRequestProcessor implements NettyRequestProcessor {
break;
}
case CONTROLLER_GET_METADATA_INFO: {
- return this.controller.getControllerMetadata();
+ final RemotingCommand response = this.controller.getControllerMetadata();
+ final GetMetaDataResponseHeader responseHeader = (GetMetaDataResponseHeader) response.readCustomHeader();
+ if (StringUtils.isNoneEmpty(responseHeader.getControllerLeaderAddress())) {
+ final String leaderAddress = responseHeader.getControllerLeaderAddress();
+ // Because the controller is proxy by namesrv, so we should replace the controllerAddress to namesrvAddress.
+ final int splitIndex = StringUtils.lastIndexOf(leaderAddress, ":");
+ final String namesrvAddress = leaderAddress.substring(0, splitIndex + 1) + this.namesrvController.getNettyServerConfig().getListenPort();
+ responseHeader.setControllerLeaderAddress(namesrvAddress);
+ }
+ return response;
+ }
+ case CONTROLLER_GET_SYNC_STATE_DATA: {
+ if (request.getBody() != null) {
+ final List<String> brokerNames = RemotingSerializable.decode(request.getBody(), List.class);
+ if (brokerNames != null && brokerNames.size() > 0) {
+ final CompletableFuture<RemotingCommand> future = this.controller.getSyncStateData(brokerNames);
+ if (future != null) {
+ return future.get(WAIT_TIMEOUT_OUT, TimeUnit.SECONDS);
+ }
+ }
+ }
}
default: {
final String error = " request type " + request.getCode() + " not supported";
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
index 941760082..773817072 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
@@ -286,6 +286,10 @@ public class AutoSwitchHAService extends DefaultHAService {
return reputFromOffset;
}
+ public List<EpochEntry> getEpochEntries() {
+ return this.epochCache.getAllEntries();
+ }
+
class AutoSwitchAcceptSocketService extends AcceptSocketService {
public AutoSwitchAcceptSocketService(int port) {
diff --git a/test/src/test/java/org/apache/rocketmq/test/autoswitchrole/AutoSwitchRoleIntegrationTest.java b/test/src/test/java/org/apache/rocketmq/test/autoswitchrole/AutoSwitchRoleIntegrationTest.java
index d2219a870..bdf23ecbd 100644
--- a/test/src/test/java/org/apache/rocketmq/test/autoswitchrole/AutoSwitchRoleIntegrationTest.java
+++ b/test/src/test/java/org/apache/rocketmq/test/autoswitchrole/AutoSwitchRoleIntegrationTest.java
@@ -58,6 +58,7 @@ public class AutoSwitchRoleIntegrationTest extends AutoSwitchRoleBase {
final NettyServerConfig serverConfig = new NettyServerConfig();
int namesrvPort = nextPort();
serverConfig.setListenPort(namesrvPort);
+ System.out.println(namesrvPort);
this.controllerConfig = buildControllerConfig("n0", peers);
this.namesrvController = new NamesrvController(new NamesrvConfig(), serverConfig, new NettyClientConfig(), controllerConfig);
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 b8bb39979..d7de1124d 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
@@ -16,6 +16,11 @@
*/
package org.apache.rocketmq.tools.admin;
+import java.io.UnsupportedEncodingException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
@@ -36,8 +41,10 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.common.protocol.body.EpochEntryCache;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.common.protocol.body.InSyncStateData;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
@@ -59,12 +66,6 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.tools.admin.api.BrokerOperatorResult;
import org.apache.rocketmq.tools.admin.api.MessageTrack;
-
-import java.io.UnsupportedEncodingException;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
import org.apache.rocketmq.tools.admin.common.AdminToolResult;
public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
@@ -731,6 +732,17 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
return this.defaultMQAdminExtImpl.getBrokerHAStatus(brokerAddr);
}
+ @Override
+ public InSyncStateData getInSyncStateData(String controllerAddress,
+ List<String> brokers) throws RemotingException, InterruptedException, MQBrokerException {
+ return this.defaultMQAdminExtImpl.getInSyncStateData(controllerAddress, brokers);
+ }
+
+ @Override public EpochEntryCache getBrokerEpochCache(
+ String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException {
+ return this.defaultMQAdminExtImpl.getBrokerEpochCache(brokerAddr);
+ }
+
@Override
public void resetMasterFlushOffset(String brokerAddr, long masterFlushOffset)
throws InterruptedException, RemotingConnectException, RemotingTimeoutException, RemotingSendRequestException, MQBrokerException {
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 7b78ead3a..302167309 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
@@ -73,8 +73,10 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.common.protocol.body.EpochEntryCache;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.common.protocol.body.InSyncStateData;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
@@ -1605,6 +1607,16 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return this.mqClientInstance.getMQClientAPIImpl().getBrokerHAStatus(brokerAddr, timeoutMillis);
}
+ @Override public InSyncStateData getInSyncStateData(String controllerAddress,
+ List<String> brokers) throws RemotingException, InterruptedException, MQBrokerException {
+ return this.mqClientInstance.getMQClientAPIImpl().getInSyncStateData(controllerAddress, brokers);
+ }
+
+ @Override public EpochEntryCache getBrokerEpochCache(
+ String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException {
+ return this.mqClientInstance.getMQClientAPIImpl().getBrokerEpochCache(brokerAddr);
+ }
+
@Override public void resetMasterFlushOffset(String brokerAddr,
long masterFlushOffset) throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
this.mqClientInstance.getMQClientAPIImpl().resetMasterFlushOffset(brokerAddr, masterFlushOffset);
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 5ed147c13..1b3492277 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
@@ -16,6 +16,11 @@
*/
package org.apache.rocketmq.tools.admin;
+import java.io.UnsupportedEncodingException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
import org.apache.rocketmq.client.MQAdmin;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
@@ -35,8 +40,10 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.common.protocol.body.EpochEntryCache;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.common.protocol.body.InSyncStateData;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
@@ -56,12 +63,6 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.tools.admin.api.BrokerOperatorResult;
import org.apache.rocketmq.tools.admin.api.MessageTrack;
-
-import java.io.UnsupportedEncodingException;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
import org.apache.rocketmq.tools.admin.common.AdminToolResult;
public interface MQAdminExt extends MQAdmin {
@@ -383,6 +384,10 @@ public interface MQAdminExt extends MQAdmin {
HARuntimeInfo getBrokerHAStatus(String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, InterruptedException, MQBrokerException;
+ InSyncStateData getInSyncStateData(String controllerAddress, List<String> brokers) throws RemotingException, InterruptedException, MQBrokerException;
+
+ EpochEntryCache getBrokerEpochCache(String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException;
+
/**
* Reset master flush offset in slave
*
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/CommandUtil.java b/tools/src/main/java/org/apache/rocketmq/tools/command/CommandUtil.java
index c2e4cb2e4..b7faee32d 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/CommandUtil.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/CommandUtil.java
@@ -135,6 +135,15 @@ public class CommandUtil {
return brokerAddressSet;
}
+ public static Set<String> fetchMasterAndSlaveAddrByBrokerName(final MQAdminExt adminExt, final String brokerName)
+ throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
+ RemotingSendRequestException, MQBrokerException {
+ ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
+ final BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
+ return new HashSet<>(brokerData.getBrokerAddrs().values());
+ }
+
+
public static Set<String> fetchBrokerNameByClusterName(final MQAdminExt adminExt, final String clusterName)
throws Exception {
ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
index bc720383b..6be8b0efc 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
@@ -19,6 +19,8 @@ package org.apache.rocketmq.tools.command;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
@@ -38,6 +40,7 @@ import org.apache.rocketmq.tools.command.broker.BrokerStatusSubCommand;
import org.apache.rocketmq.tools.command.broker.CleanExpiredCQSubCommand;
import org.apache.rocketmq.tools.command.broker.CleanUnusedTopicCommand;
import org.apache.rocketmq.tools.command.broker.GetBrokerConfigCommand;
+import org.apache.rocketmq.tools.command.broker.GetBrokerEpochCacheCommand;
import org.apache.rocketmq.tools.command.broker.ResetMasterFlushOffsetSubCommand;
import org.apache.rocketmq.tools.command.broker.SendMsgStatusCommand;
import org.apache.rocketmq.tools.command.broker.UpdateBrokerConfigSubCommand;
@@ -58,6 +61,7 @@ import org.apache.rocketmq.tools.command.export.ExportConfigsCommand;
import org.apache.rocketmq.tools.command.export.ExportMetadataCommand;
import org.apache.rocketmq.tools.command.export.ExportMetricsCommand;
import org.apache.rocketmq.tools.command.ha.HAStatusSubCommand;
+import org.apache.rocketmq.tools.command.ha.SyncStateSetCommand;
import org.apache.rocketmq.tools.command.message.CheckMsgSendRTCommand;
import org.apache.rocketmq.tools.command.message.ConsumeMessageCommand;
import org.apache.rocketmq.tools.command.message.PrintMessageByQueueCommand;
@@ -92,9 +96,6 @@ import org.apache.rocketmq.tools.command.topic.UpdateTopicPermSubCommand;
import org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.List;
-
public class MQAdminStartup {
protected static List<SubCommand> subCommandList = new ArrayList<SubCommand>();
@@ -246,6 +247,9 @@ public class MQAdminStartup {
initCommand(new ExportMetricsCommand());
initCommand(new HAStatusSubCommand());
+
+ initCommand(new SyncStateSetCommand());
+ initCommand(new GetBrokerEpochCacheCommand());
}
private static void initLogback() throws JoranException {
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerEpochCacheCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerEpochCacheCommand.java
new file mode 100644
index 000000000..28f181328
--- /dev/null
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerEpochCacheCommand.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.tools.command.broker;
+
+import java.util.List;
+import java.util.Set;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.rocketmq.common.EpochEntry;
+import org.apache.rocketmq.common.protocol.body.EpochEntryCache;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.srvutil.ServerUtil;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.command.CommandUtil;
+import org.apache.rocketmq.tools.command.SubCommand;
+import org.apache.rocketmq.tools.command.SubCommandException;
+
+public class GetBrokerEpochCacheCommand implements SubCommand {
+ @Override public String commandName() {
+ return "getBrokerEpochCache";
+ }
+
+ @Override public String commandDesc() {
+ return "Fetch broker epoch entries cache";
+ }
+
+ @Override public Options buildCommandlineOptions(Options options) {
+ Option opt = new Option("c", "clusterName", true, "which cluster");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("b", "brokerName", true, "which broker to fetch");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("f", "follow", true, "the interval(second) of get info");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ return options;
+ }
+
+ @Override public void execute(CommandLine commandLine, Options options,
+ RPCHook rpcHook) throws SubCommandException {
+ DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
+ defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
+
+ try {
+ if (commandLine.hasOption('f')) {
+ String flushSecondStr = commandLine.getOptionValue('f');
+ int flushSecond = 3;
+ if (flushSecondStr != null && !flushSecondStr.trim().equals("")) {
+ flushSecond = Integer.parseInt(flushSecondStr);
+ }
+
+ defaultMQAdminExt.start();
+
+ while (true) {
+ this.innerExec(commandLine, options, defaultMQAdminExt);
+ Thread.sleep(flushSecond * 1000);
+ }
+ } else {
+ defaultMQAdminExt.start();
+
+ this.innerExec(commandLine, options, defaultMQAdminExt);
+ }
+ } catch (Exception e) {
+ throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
+ } finally {
+ defaultMQAdminExt.shutdown();
+ }
+ }
+
+ private void innerExec(CommandLine commandLine, Options options,
+ DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+ if (commandLine.hasOption('b')) {
+ String brokerName = commandLine.getOptionValue('b').trim();
+ final Set<String> brokers = CommandUtil.fetchMasterAndSlaveAddrByBrokerName(defaultMQAdminExt, brokerName);
+ printData(brokers, defaultMQAdminExt);
+ } else if (commandLine.hasOption('c')) {
+ String clusterName = commandLine.getOptionValue('c').trim();
+ Set<String> brokers = CommandUtil.fetchMasterAndSlaveAddrByClusterName(defaultMQAdminExt, clusterName);
+ printData(brokers, defaultMQAdminExt);
+ } else {
+ ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
+ }
+ }
+
+ private void printData(Set<String> brokers, DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+ for (String brokerAddr : brokers) {
+ final EpochEntryCache epochCache = defaultMQAdminExt.getBrokerEpochCache(brokerAddr);
+ System.out.printf("\n#clusterName\t%s\n#brokerName\t%s\n#brokerAddr\t%s\n#brokerId\t%d",
+ epochCache.getClusterName(), epochCache.getBrokerName(), brokerAddr, epochCache.getBrokerId());
+ final List<EpochEntry> epochList = epochCache.getEpochList();
+ for (int i = 0; i < epochList.size(); i++) {
+ final EpochEntry epochEntry = epochList.get(i);
+ if (i == epochList.size() - 1) {
+ epochEntry.setEndOffset(epochCache.getMaxOffset());
+ }
+ System.out.printf("\n#Epoch: %s", epochEntry.toString());
+ }
+ }
+ }
+}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/ha/SyncStateSetCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/ha/SyncStateSetCommand.java
new file mode 100644
index 000000000..52b48f871
--- /dev/null
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/ha/SyncStateSetCommand.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.tools.command.ha;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.rocketmq.common.protocol.body.InSyncStateData;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.srvutil.ServerUtil;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.command.CommandUtil;
+import org.apache.rocketmq.tools.command.SubCommand;
+import org.apache.rocketmq.tools.command.SubCommandException;
+
+public class SyncStateSetCommand implements SubCommand {
+ @Override
+ public String commandName() {
+ return "getSyncStateSet";
+ }
+
+ @Override
+ public String commandDesc() {
+ return "Fetch syncStateSet for target brokers";
+ }
+
+ @Override
+ public Options buildCommandlineOptions(Options options) {
+ Option opt = new Option("a", "controllerAddress", true, "the address of controller");
+ opt.setRequired(true);
+ options.addOption(opt);
+
+ opt = new Option("c", "clusterName", true, "which cluster");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("b", "brokerName", true, "which broker to fetch");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("f", "follow", true, "the interval(second) of get info");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ return options;
+ }
+
+ @Override
+ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
+ DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
+ defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
+
+ try {
+ if (commandLine.hasOption('f')) {
+ String flushSecondStr = commandLine.getOptionValue('f');
+ int flushSecond = 3;
+ if (flushSecondStr != null && !flushSecondStr.trim().equals("")) {
+ flushSecond = Integer.parseInt(flushSecondStr);
+ }
+
+ defaultMQAdminExt.start();
+
+ while (true) {
+ this.innerExec(commandLine, options, defaultMQAdminExt);
+ Thread.sleep(flushSecond * 1000);
+ }
+ } else {
+ defaultMQAdminExt.start();
+
+ this.innerExec(commandLine, options, defaultMQAdminExt);
+ }
+ } catch (Exception e) {
+ throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
+ } finally {
+ defaultMQAdminExt.shutdown();
+ }
+ }
+
+ private void innerExec(CommandLine commandLine, Options options,
+ DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+ String controllerAddress = commandLine.getOptionValue('a').trim();
+ if (commandLine.hasOption('b')) {
+ String brokerName = commandLine.getOptionValue('b').trim();
+ final ArrayList<String> brokers = new ArrayList<>();
+ brokers.add(brokerName);
+ printData(controllerAddress, brokers, defaultMQAdminExt);
+ } else if (commandLine.hasOption('c')) {
+ String clusterName = commandLine.getOptionValue('c').trim();
+ Set<String> brokerNames = CommandUtil.fetchBrokerNameByClusterName(defaultMQAdminExt, clusterName);
+ printData(controllerAddress, new ArrayList<>(brokerNames), defaultMQAdminExt);
+ } else {
+ ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
+ }
+ }
+
+ private void printData(String controllerAddress, List<String> brokerNames,
+ DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+ if (brokerNames.size() > 0) {
+ final InSyncStateData syncStateData = defaultMQAdminExt.getInSyncStateData(controllerAddress, brokerNames);
+ final Map<String, InSyncStateData.InSyncState> syncTable = syncStateData.getInSyncTable();
+ for (Map.Entry<String, InSyncStateData.InSyncState> next : syncTable.entrySet()) {
+ final List<InSyncStateData.InSyncMember> 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",
+ 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", member.toString());
+ }
+ }
+ }
+ }
+}