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 c425dba291..7719b0ae68 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 @@ -79,6 +79,7 @@ import org.apache.rocketmq.common.namesrv.TopAddressing; import org.apache.rocketmq.common.protocol.NamespaceUtil; import org.apache.rocketmq.common.protocol.RequestCode; import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.common.protocol.body.BrokerMemberGroup; import org.apache.rocketmq.common.protocol.body.BrokerStatsData; import org.apache.rocketmq.common.protocol.body.CheckClientRequestBody; import org.apache.rocketmq.common.protocol.body.ClusterAclVersionInfo; @@ -185,6 +186,8 @@ 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.ElectMasterRequestHeader; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader; 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; @@ -3026,4 +3029,32 @@ public class MQClientAPIImpl implements NameServerUpdateCallback { throw new MQClientException(errResponse.getCode(), errResponse.getRemark()); } } + + public ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, String brokerName, + String brokerAddr) throws MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, RemotingCommandException { + + //get controller leader address + final GetMetaDataResponseHeader controllerMetaData = this.getControllerMetaData(controllerAddr); + assert controllerMetaData != null; + assert controllerMetaData.getControllerLeaderAddress() != null; + final String leaderAddress = controllerMetaData.getControllerLeaderAddress(); + ElectMasterRequestHeader electRequestHeader = new ElectMasterRequestHeader(clusterName, brokerName, brokerAddr); + + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_ELECT_MASTER, electRequestHeader); + final RemotingCommand response = this.remotingClient.invokeSync(leaderAddress, request, 3000); + assert response != null; + switch (response.getCode()) { + case ResponseCode.SUCCESS: { + BrokerMemberGroup brokerMemberGroup = RemotingSerializable.decode(response.getBody(), BrokerMemberGroup.class); + ElectMasterResponseHeader responseHeader = (ElectMasterResponseHeader) response.decodeCommandCustomHeader(ElectMasterResponseHeader.class); + if (null != responseHeader) { + responseHeader.setBrokerMemberGroup(brokerMemberGroup); + } + return responseHeader; + } + default: + break; + } + throw new MQBrokerException(response.getCode(), response.getRemark()); + } } diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/controller/ElectMasterRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/controller/ElectMasterRequestHeader.java index d96355e660..13146a4fbe 100644 --- a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/controller/ElectMasterRequestHeader.java +++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/controller/ElectMasterRequestHeader.java @@ -17,11 +17,20 @@ package org.apache.rocketmq.common.protocol.header.namesrv.controller; import org.apache.rocketmq.remoting.CommandCustomHeader; +import org.apache.rocketmq.remoting.annotation.CFNotNull; import org.apache.rocketmq.remoting.exception.RemotingCommandException; public class ElectMasterRequestHeader implements CommandCustomHeader { + + @CFNotNull + private String clusterName; + + @CFNotNull private String brokerName; + @CFNotNull + private String brokerAddress; + public ElectMasterRequestHeader() { } @@ -29,6 +38,12 @@ public class ElectMasterRequestHeader implements CommandCustomHeader { this.brokerName = brokerName; } + public ElectMasterRequestHeader(String clusterName, String brokerName, String brokerAddress) { + this.clusterName = clusterName; + this.brokerName = brokerName; + this.brokerAddress = brokerAddress; + } + public String getBrokerName() { return brokerName; } @@ -37,10 +52,28 @@ public class ElectMasterRequestHeader implements CommandCustomHeader { this.brokerName = brokerName; } + public String getBrokerAddress() { + return brokerAddress; + } + + public void setBrokerAddress(String brokerAddress) { + this.brokerAddress = brokerAddress; + } + + public String getClusterName() { + return clusterName; + } + + public void setClusterName(String clusterName) { + this.clusterName = clusterName; + } + @Override public String toString() { return "ElectMasterRequestHeader{" + + "clusterName='" + clusterName + '\'' + "brokerName='" + brokerName + '\'' + + "brokerAddress='" + brokerAddress + '\'' + '}'; } 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 a53c7ca9d7..c63ca0bf1b 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 @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Set; import java.util.function.BiPredicate; import java.util.function.Predicate; + import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.constant.LoggerName; @@ -67,8 +68,8 @@ public class ReplicasInfoManager { } public ControllerResult alterSyncStateSet( - final AlterSyncStateSetRequestHeader request, final SyncStateSet syncStateSet, - final BiPredicate brokerAlivePredicate) { + final AlterSyncStateSetRequestHeader request, final SyncStateSet syncStateSet, + final BiPredicate brokerAlivePredicate) { final String brokerName = request.getBrokerName(); final ControllerResult result = new ControllerResult<>(new AlterSyncStateSetResponseHeader()); final AlterSyncStateSetResponseHeader response = result.getResponse(); @@ -90,7 +91,7 @@ public class ReplicasInfoManager { // Check master if (!syncStateInfo.getMasterAddress().equals(request.getMasterAddress())) { String err = String.format("Rejecting alter syncStateSet request because the current leader is:{%s}, not {%s}", - syncStateInfo.getMasterAddress(), request.getMasterAddress()); + syncStateInfo.getMasterAddress(), request.getMasterAddress()); log.error("{}", err); result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_MASTER, err); return result; @@ -99,7 +100,7 @@ public class ReplicasInfoManager { // Check master epoch if (request.getMasterEpoch() != syncStateInfo.getMasterEpoch()) { String err = String.format("Rejecting alter syncStateSet request because the current master epoch is:{%d}, not {%d}", - syncStateInfo.getMasterEpoch(), request.getMasterEpoch()); + syncStateInfo.getMasterEpoch(), request.getMasterEpoch()); log.error("{}", err); result.setCodeAndRemark(ResponseCode.CONTROLLER_FENCED_MASTER_EPOCH, err); return result; @@ -108,7 +109,7 @@ public class ReplicasInfoManager { // Check syncStateSet epoch if (syncStateSet.getSyncStateSetEpoch() != syncStateInfo.getSyncStateSetEpoch()) { String err = String.format("Rejecting alter syncStateSet request because the current syncStateSet epoch is:{%d}, not {%d}", - syncStateInfo.getSyncStateSetEpoch(), syncStateSet.getSyncStateSetEpoch()); + syncStateInfo.getSyncStateSetEpoch(), syncStateSet.getSyncStateSetEpoch()); log.error("{}", err); result.setCodeAndRemark(ResponseCode.CONTROLLER_FENCED_SYNC_STATE_SET_EPOCH, err); return result; @@ -150,8 +151,9 @@ public class ReplicasInfoManager { } public ControllerResult electMaster( - final ElectMasterRequestHeader request, final BiPredicate brokerAlivePredicate) { + final ElectMasterRequestHeader request, final BiPredicate brokerAlivePredicate) { final String brokerName = request.getBrokerName(); + final String assignBrokerAddress = request.getBrokerAddress(); final ControllerResult result = new ControllerResult<>(new ElectMasterResponseHeader()); if (isContainsBroker(brokerName)) { final SyncStateInfo syncStateInfo = this.syncStateSetInfoTable.get(brokerName); @@ -160,16 +162,26 @@ public class ReplicasInfoManager { // First, check whether the master is still active final String oldMaster = syncStateInfo.getMasterAddress(); if (StringUtils.isNoneEmpty(oldMaster) && brokerAlivePredicate.test(brokerInfo.getClusterName(), oldMaster)) { - String err = String.format("The old master %s is still alive, not need to elect new master for broker %s", oldMaster, brokerInfo.getBrokerName()); - log.warn("{}", err); - result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_REQUEST, err); - return result; + + if (StringUtils.isBlank(assignBrokerAddress)) { + String err = String.format("The old master %s is still alive, no need to elect new master for broker %s", oldMaster, brokerInfo.getBrokerName()); + log.warn("{}", err); + result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_REQUEST, err); + return result; + } + + if (StringUtils.equals(oldMaster, assignBrokerAddress)) { + String err = String.format("The Re-elect master is the same as the old master %s which is still alive, no need to elect new master for broker %s", oldMaster, brokerInfo.getBrokerName()); + log.warn("{}", err); + result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_REQUEST, err); + return result; + } } // Try elect a master in syncStateSet if (syncStateSet.size() > 1) { - boolean electSuccess = tryElectMaster(result, brokerName, syncStateSet, candidate -> - !candidate.equals(syncStateInfo.getMasterAddress()) && brokerAlivePredicate.test(brokerInfo.getClusterName(), candidate)); + boolean electSuccess = tryElectMaster(result, brokerName, assignBrokerAddress, syncStateSet, candidate -> + !candidate.equals(syncStateInfo.getMasterAddress()) && brokerAlivePredicate.test(brokerInfo.getClusterName(), candidate)); if (electSuccess) { return result; } @@ -177,8 +189,8 @@ public class ReplicasInfoManager { // Try elect a master in lagging replicas if enableElectUncleanMaster = true if (controllerConfig.isEnableElectUncleanMaster()) { - boolean electSuccess = tryElectMaster(result, brokerName, brokerInfo.getAllBroker(), candidate -> - !candidate.equals(syncStateInfo.getMasterAddress()) && brokerAlivePredicate.test(brokerInfo.getClusterName(), candidate)); + boolean electSuccess = tryElectMaster(result, brokerName, assignBrokerAddress, brokerInfo.getAllBroker(), candidate -> + !candidate.equals(syncStateInfo.getMasterAddress()) && brokerAlivePredicate.test(brokerInfo.getClusterName(), candidate)); if (electSuccess) { return result; } @@ -202,9 +214,28 @@ public class ReplicasInfoManager { * @return true if elect success */ private boolean tryElectMaster(final ControllerResult result, final String brokerName, - final Set candidates, final Predicate filter) { + final String assignBrokerAddress, final Set candidates, final Predicate filter) { final int masterEpoch = this.syncStateSetInfoTable.get(brokerName).getMasterEpoch(); final int syncStateSetEpoch = this.syncStateSetInfoTable.get(brokerName).getSyncStateSetEpoch(); + + //Re-elect the assigned broker as master + if (StringUtils.isNotBlank(assignBrokerAddress) && filter.test(assignBrokerAddress)) { + final ElectMasterResponseHeader response = result.getResponse(); + response.setNewMasterAddress(assignBrokerAddress); + response.setMasterEpoch(masterEpoch + 1); + response.setSyncStateSetEpoch(syncStateSetEpoch); + BrokerMemberGroup brokerMemberGroup = buildBrokerMemberGroup(brokerName); + if (null != brokerMemberGroup) { + response.setBrokerMemberGroup(brokerMemberGroup); + result.setBody(brokerMemberGroup.encode()); + } + final ElectMasterEvent event = new ElectMasterEvent(brokerName, assignBrokerAddress); + result.addEvent(event); + return true; + } else if (StringUtils.isNotBlank(assignBrokerAddress) && !filter.test(assignBrokerAddress)) { + return false; + } + for (final String candidate : candidates) { if (filter.test(candidate)) { final ElectMasterResponseHeader response = result.getResponse(); @@ -213,6 +244,7 @@ public class ReplicasInfoManager { response.setSyncStateSetEpoch(syncStateSetEpoch); BrokerMemberGroup brokerMemberGroup = buildBrokerMemberGroup(brokerName); if (null != brokerMemberGroup) { + response.setBrokerMemberGroup(brokerMemberGroup); result.setBody(brokerMemberGroup.encode()); } final ElectMasterEvent event = new ElectMasterEvent(brokerName, candidate); @@ -237,7 +269,7 @@ public class ReplicasInfoManager { } public ControllerResult registerBroker( - final RegisterBrokerToControllerRequestHeader request) { + final RegisterBrokerToControllerRequestHeader request) { final String brokerName = request.getBrokerName(); final String brokerAddress = request.getBrokerAddress(); final ControllerResult result = new ControllerResult<>(new RegisterBrokerToControllerResponseHeader()); 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 12824d7c36..5b583b493b 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,21 +17,25 @@ package org.apache.rocketmq.controller.processor; import io.netty.channel.ChannelHandlerContext; + import java.io.UnsupportedEncodingException; import java.util.List; import java.util.Properties; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.common.MixAll; 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.BrokerHeartbeatRequestHeader; import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetRequestHeader; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterRequestHeader; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoRequestHeader; import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerRequestHeader; import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerResponseHeader; -import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterRequestHeader; -import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoRequestHeader; import org.apache.rocketmq.controller.BrokerHeartbeatManager; import org.apache.rocketmq.controller.ControllerManager; import org.apache.rocketmq.logging.InternalLogger; @@ -69,9 +73,9 @@ public class ControllerRequestProcessor implements NettyRequestProcessor { public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception { if (ctx != null) { log.debug("Receive request, {} {} {}", - request.getCode(), - RemotingHelper.parseChannelRemoteAddr(ctx.channel()), - request); + request.getCode(), + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), + request); } switch (request.getCode()) { case CONTROLLER_ALTER_SYNC_STATE_SET: { @@ -84,10 +88,21 @@ public class ControllerRequestProcessor implements NettyRequestProcessor { break; } case CONTROLLER_ELECT_MASTER: { - final ElectMasterRequestHeader controllerRequest = (ElectMasterRequestHeader) request.decodeCommandCustomHeader(ElectMasterRequestHeader.class); - final CompletableFuture future = this.controllerManager.getController().electMaster(controllerRequest); + final ElectMasterRequestHeader electMasterRequest = (ElectMasterRequestHeader) request.decodeCommandCustomHeader(ElectMasterRequestHeader.class); + final CompletableFuture future = this.controllerManager.getController().electMaster(electMasterRequest); if (future != null) { - return future.get(WAIT_TIMEOUT_OUT, TimeUnit.SECONDS); + final RemotingCommand response = future.get(WAIT_TIMEOUT_OUT, TimeUnit.SECONDS); + final ElectMasterResponseHeader responseHeader = (ElectMasterResponseHeader) response.readCustomHeader(); + + if (null != responseHeader) { + if (StringUtils.isNotEmpty(responseHeader.getNewMasterAddress())) { + heartbeatManager.changeBrokerMetadata(electMasterRequest.getClusterName(), responseHeader.getNewMasterAddress(), MixAll.MASTER_ID); + } + if (this.controllerManager.getControllerConfig().isNotifyBrokerRoleChanged()) { + this.controllerManager.notifyBrokerRoleChanged(responseHeader, electMasterRequest.getClusterName()); + } + } + return response; } break; } @@ -99,7 +114,7 @@ public class ControllerRequestProcessor implements NettyRequestProcessor { final RegisterBrokerToControllerResponseHeader responseHeader = (RegisterBrokerToControllerResponseHeader) response.readCustomHeader(); if (responseHeader != null && responseHeader.getBrokerId() >= 0) { this.heartbeatManager.registerBroker(controllerRequest.getClusterName(), controllerRequest.getBrokerName(), controllerRequest.getBrokerAddress(), - responseHeader.getBrokerId(), controllerRequest.getHeartbeatTimeoutMillis(), ctx.channel()); + responseHeader.getBrokerId(), controllerRequest.getHeartbeatTimeoutMillis(), ctx.channel()); } return response; } diff --git a/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/manager/ReplicasInfoManagerTest.java b/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/manager/ReplicasInfoManagerTest.java index 09ed9fc70e..ed47be2df8 100644 --- a/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/manager/ReplicasInfoManagerTest.java +++ b/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/manager/ReplicasInfoManagerTest.java @@ -20,6 +20,7 @@ import java.util.HashSet; import java.util.List; import java.util.Set; import org.apache.rocketmq.common.ControllerConfig; +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.AlterSyncStateSetResponseHeader; @@ -53,7 +54,8 @@ public class ReplicasInfoManagerTest { this.replicasInfoManager = new ReplicasInfoManager(config); } - public boolean registerNewBroker(String clusterName, String brokerName, String brokerAddress, boolean isFirstRegisteredBroker) { + public boolean registerNewBroker(String clusterName, String brokerName, String brokerAddress, + boolean isFirstRegisteredBroker) { // Register new broker final RegisterBrokerToControllerRequestHeader registerRequest = new RegisterBrokerToControllerRequestHeader(clusterName, brokerName, brokerAddress); @@ -72,7 +74,8 @@ public class ReplicasInfoManagerTest { return true; } - private boolean alterNewInSyncSet(String brokerName, String masterAddress, int masterEpoch, Set newSyncStateSet, int syncStateSetEpoch) { + private boolean alterNewInSyncSet(String brokerName, String masterAddress, int masterEpoch, + Set newSyncStateSet, int syncStateSetEpoch) { final AlterSyncStateSetRequestHeader alterRequest = new AlterSyncStateSetRequestHeader(brokerName, masterAddress, masterEpoch); final ControllerResult result = this.replicasInfoManager.alterSyncStateSet(alterRequest, new SyncStateSet(newSyncStateSet, syncStateSetEpoch), (va1, va2) -> true); @@ -113,6 +116,29 @@ public class ReplicasInfoManagerTest { assertEquals(response.getMasterEpoch(), 2); assertFalse(response.getNewMasterAddress().isEmpty()); assertNotEquals(response.getNewMasterAddress(), "127.0.0.1:9000"); + + final Set brokerSet = new HashSet<>(); + brokerSet.add("127.0.0.1:9000"); + brokerSet.add("127.0.0.1:9001"); + brokerSet.add("127.0.0.1:9002"); + final ElectMasterRequestHeader assignRequest = new ElectMasterRequestHeader("cluster1","broker1", "127.0.0.1:9000"); + final ControllerResult cResult1 = this.replicasInfoManager.electMaster(assignRequest, (clusterName, brokerAddress) -> brokerAddress.contains("127.0.0.1:9000")); + assertEquals( cResult1.getResponseCode(), ResponseCode.CONTROLLER_INVALID_REQUEST); + + + final ElectMasterRequestHeader assignRequest1 = new ElectMasterRequestHeader("cluster1","broker1", "127.0.0.1:9001"); + final ControllerResult cResult2 = this.replicasInfoManager.electMaster(assignRequest1, (clusterName, brokerAddress) -> brokerAddress.equals("127.0.0.1:9000")); + assertEquals( cResult2.getResponseCode(), ResponseCode.CONTROLLER_MASTER_NOT_AVAILABLE); + + final ElectMasterRequestHeader assignRequest2 = new ElectMasterRequestHeader("cluster1","broker1", "127.0.0.1:9001"); + final ControllerResult cResult3 = this.replicasInfoManager.electMaster(assignRequest2, (clusterName, brokerAddress) -> !brokerAddress.equals("127.0.0.1:9000")); + assertEquals( cResult3.getResponseCode(), ResponseCode.SUCCESS); + final ElectMasterResponseHeader response3 = cResult3.getResponse(); + assertEquals(response3.getNewMasterAddress(),"127.0.0.1:9001"); + assertEquals(response.getMasterEpoch(), 2); + assertFalse(response.getNewMasterAddress().isEmpty()); + assertNotEquals(response.getNewMasterAddress(), "127.0.0.1:9000"); + } @Test 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 3356f94f76..0b550b56b5 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 @@ -53,6 +53,7 @@ import org.apache.rocketmq.common.protocol.body.QueueTimeSpan; import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper; import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper; import org.apache.rocketmq.common.protocol.body.TopicList; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader; import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetMetaDataResponseHeader; import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData; import org.apache.rocketmq.common.protocol.route.TopicRouteData; @@ -208,7 +209,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { @Override public void updateGlobalWhiteAddrConfig(String addr, - String globalWhiteAddrs, String aclFileFullPath) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { + String globalWhiteAddrs, + String aclFileFullPath) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { defaultMQAdminExtImpl.updateGlobalWhiteAddrConfig(addr, globalWhiteAddrs, aclFileFullPath); } @@ -333,7 +335,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { } @Override - public ProducerTableInfo getAllProducerInfo(final String brokerAddr) throws RemotingException, MQClientException, InterruptedException, MQBrokerException { + public ProducerTableInfo getAllProducerInfo( + final String brokerAddr) throws RemotingException, MQClientException, InterruptedException, MQBrokerException { return defaultMQAdminExtImpl.getAllProducerInfo(brokerAddr); } @@ -534,7 +537,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { } @Override - public boolean deleteExpiredCommitLogByAddr(String addr) throws RemotingConnectException, RemotingSendRequestException, + public boolean deleteExpiredCommitLogByAddr( + String addr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException { return defaultMQAdminExtImpl.deleteExpiredCommitLogByAddr(addr); } @@ -787,7 +791,6 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { this.defaultMQAdminExtImpl.resetMasterFlushOffset(brokerAddr, masterFlushOffset); } - public QueryResult queryMessageByUniqKey(String topic, String key, int maxNum, long begin, long end) throws MQClientException, InterruptedException { @@ -808,7 +811,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { } @Override - public Map getControllerConfig(List controllerServers) throws InterruptedException, RemotingTimeoutException, + public Map getControllerConfig( + List controllerServers) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException, UnsupportedEncodingException { return this.defaultMQAdminExtImpl.getControllerConfig(controllerServers); @@ -819,4 +823,10 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { List controllers) throws InterruptedException, RemotingConnectException, UnsupportedEncodingException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException { this.defaultMQAdminExtImpl.updateControllerConfig(properties, controllers); } + + @Override + public ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, + String brokerName, String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException { + return this.defaultMQAdminExtImpl.electMaster(controllerAddr, clusterName, brokerName, brokerAddr); + } } 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 ee28c876eb..3cea455a2d 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 @@ -60,6 +60,7 @@ import org.apache.rocketmq.common.admin.TopicStatsTable; import org.apache.rocketmq.common.help.FAQUrl; import org.apache.rocketmq.common.protocol.body.ClusterAclVersionInfo; import org.apache.rocketmq.common.protocol.body.ProducerTableInfo; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.common.message.MessageClientExt; import org.apache.rocketmq.common.message.MessageConst; @@ -1793,6 +1794,12 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner { this.mqClientInstance.getMQClientAPIImpl().resetMasterFlushOffset(brokerAddr, masterFlushOffset); } + @Override + public ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, + String brokerName, String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException { + return this.mqClientInstance.getMQClientAPIImpl().electMaster(controllerAddr, clusterName, brokerName, brokerAddr); + } + @Override public GroupForbidden updateAndGetGroupReadForbidden(String brokerAddr, String groupName, String topicName, Boolean readable) throws RemotingException, InterruptedException, MQBrokerException { @@ -1823,7 +1830,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner { return this.mqClientInstance.getMQClientAPIImpl().getControllerConfig(controllerServers, timeoutMillis); } - @Override public void updateControllerConfig(Properties properties, + @Override + public void updateControllerConfig(Properties properties, List controllers) throws InterruptedException, RemotingConnectException, UnsupportedEncodingException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException { this.mqClientInstance.getMQClientAPIImpl().updateControllerConfig(properties, controllers, timeoutMillis); 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 9fe6ea4fee..857f159b5e 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 @@ -52,6 +52,7 @@ import org.apache.rocketmq.common.protocol.body.QueueTimeSpan; import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper; import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper; import org.apache.rocketmq.common.protocol.body.TopicList; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader; import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetMetaDataResponseHeader; import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData; import org.apache.rocketmq.common.protocol.route.TopicRouteData; @@ -99,7 +100,8 @@ public interface MQAdminExt extends MQAdmin { final String globalWhiteAddrs) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; - void updateGlobalWhiteAddrConfig(final String addr, final String globalWhiteAddrs, String aclFileFullPath) throws RemotingException, MQBrokerException, + void updateGlobalWhiteAddrConfig(final String addr, final String globalWhiteAddrs, + String aclFileFullPath) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; ClusterAclVersionInfo examineBrokerClusterAclVersionInfo( @@ -113,14 +115,16 @@ public interface MQAdminExt extends MQAdmin { final SubscriptionGroupConfig config) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; - SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, final String group) throws InterruptedException, RemotingException, MQClientException, MQBrokerException; + SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, + final String group) throws InterruptedException, RemotingException, MQClientException, MQBrokerException; TopicStatsTable examineTopicStats( final String topic) throws RemotingException, MQClientException, InterruptedException, MQBrokerException; - TopicStatsTable examineTopicStats(String brokerAddr, final String topic) throws RemotingException, MQClientException, InterruptedException, - MQBrokerException; + TopicStatsTable examineTopicStats(String brokerAddr, + final String topic) throws RemotingException, MQClientException, InterruptedException, + MQBrokerException; AdminToolResult examineTopicStatsConcurrent(String topic); @@ -162,7 +166,7 @@ public interface MQAdminExt extends MQAdmin { MQClientException, InterruptedException, MQBrokerException; ProducerTableInfo getAllProducerInfo(final String brokerAddr) throws RemotingException, - MQClientException, InterruptedException, MQBrokerException; + MQClientException, InterruptedException, MQBrokerException; List getNameServerAddressList(); @@ -170,7 +174,7 @@ public interface MQAdminExt extends MQAdmin { RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQClientException; int addWritePermOfBroker(final String namesrvAddr, String brokerName) throws RemotingCommandException, - RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQClientException; + RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQClientException; void putKVConfig(final String namespace, final String key, final String value); @@ -200,7 +204,8 @@ public interface MQAdminExt extends MQAdmin { void deleteSubscriptionGroup(final String addr, String groupName) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; - void deleteSubscriptionGroup(final String addr, String groupName, boolean removeOffset) throws RemotingException, MQBrokerException, + void deleteSubscriptionGroup(final String addr, String groupName, + boolean removeOffset) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; void createAndUpdateKvConfig(String namespace, String key, @@ -233,11 +238,13 @@ public interface MQAdminExt extends MQAdmin { GroupList queryTopicConsumeByWho(final String topic) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException, RemotingException, MQClientException; - TopicList queryTopicsByConsumer(final String group) throws InterruptedException, MQBrokerException, RemotingException, MQClientException; + TopicList queryTopicsByConsumer( + final String group) throws InterruptedException, MQBrokerException, RemotingException, MQClientException; AdminToolResult queryTopicsByConsumerConcurrent(final String group); - SubscriptionData querySubscription(final String group, final String topic) throws InterruptedException, MQBrokerException, RemotingException, MQClientException; + SubscriptionData querySubscription(final String group, + final String topic) throws InterruptedException, MQBrokerException, RemotingException, MQClientException; List queryConsumeTimeSpan(final String topic, final String group) throws InterruptedException, MQBrokerException, @@ -266,7 +273,8 @@ public interface MQAdminExt extends MQAdmin { ConsumerRunningInfo getConsumerRunningInfo(final String consumerGroup, final String clientId, final boolean jstack) throws RemotingException, MQClientException, InterruptedException; - ConsumerRunningInfo getConsumerRunningInfo(final String consumerGroup, final String clientId, final boolean jstack, final boolean metrics) + ConsumerRunningInfo getConsumerRunningInfo(final String consumerGroup, final String clientId, final boolean jstack, + final boolean metrics) throws RemotingException, MQClientException, InterruptedException; ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, @@ -311,7 +319,6 @@ public interface MQAdminExt extends MQAdmin { long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException; - TopicConfigSerializeWrapper getAllTopicConfig(final String brokerAddr, long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException; @@ -368,8 +375,8 @@ public interface MQAdminExt extends MQAdmin { boolean resumeCheckHalfMessage(final String topic, final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException; - void setMessageRequestMode(final String brokerAddr, final String topic, final String consumerGroup, final - MessageRequestMode mode, final int popWorkGroupSize, final long timeoutMillis) + void setMessageRequestMode(final String brokerAddr, final String topic, final String consumerGroup, + final MessageRequestMode mode, final int popWorkGroupSize, final long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException; @@ -383,15 +390,16 @@ public interface MQAdminExt extends MQAdmin { throws RemotingException, InterruptedException, MQBrokerException; TopicConfig examineTopicConfig(final String addr, - final String topic) throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException; + final String topic) throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException; - void createStaticTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig, final TopicQueueMappingDetail mappingDetail, final boolean force) throws RemotingException, InterruptedException, MQBrokerException; + void createStaticTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig, + final TopicQueueMappingDetail mappingDetail, + final boolean force) throws RemotingException, InterruptedException, MQBrokerException; GroupForbidden updateAndGetGroupReadForbidden(String brokerAddr, String groupName, String topicName, Boolean readable) throws RemotingException, InterruptedException, MQBrokerException; - MessageExt queryMessage(String clusterName, String topic, String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; @@ -399,21 +407,24 @@ public interface MQAdminExt extends MQAdmin { HARuntimeInfo getBrokerHAStatus(String brokerAddr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException; - InSyncStateData getInSyncStateData(String controllerAddress, List brokers) throws RemotingException, InterruptedException, MQBrokerException; + InSyncStateData getInSyncStateData(String controllerAddress, + List brokers) throws RemotingException, InterruptedException, MQBrokerException; - EpochEntryCache getBrokerEpochCache(String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException; + EpochEntryCache getBrokerEpochCache( + String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException; + + GetMetaDataResponseHeader getControllerMetaData( + String controllerAddr) throws RemotingException, InterruptedException, MQBrokerException; - GetMetaDataResponseHeader getControllerMetaData(String controllerAddr) throws RemotingException, InterruptedException, MQBrokerException; /** * Reset master flush offset in slave * - * @param brokerAddr slave broker address + * @param brokerAddr slave broker address * @param masterFlushOffset master flush offset */ void resetMasterFlushOffset(String brokerAddr, long masterFlushOffset) throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException; - /** * Get controller config. *
@@ -421,7 +432,8 @@ public interface MQAdminExt extends MQAdmin { * * @return The fetched controller config */ - Map getControllerConfig(List controllerServers) throws InterruptedException, RemotingTimeoutException, + Map getControllerConfig( + List controllerServers) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException, UnsupportedEncodingException; /** @@ -433,4 +445,18 @@ public interface MQAdminExt extends MQAdmin { final List controllers) throws InterruptedException, RemotingConnectException, UnsupportedEncodingException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException; + /** + * manual trigger broker elect master + * + * @param controllerAddr controller address + * @param brokerName broker name + * @param brokerAddr broker address + * @return + * @throws RemotingException + * @throws InterruptedException + * @throws MQBrokerException + */ + ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, String brokerName, + String brokerAddr) + throws RemotingException, InterruptedException, MQBrokerException; } 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 fc9226d8a4..2673c6747a 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 @@ -63,6 +63,7 @@ import org.apache.rocketmq.tools.command.container.RemoveBrokerSubCommand; import org.apache.rocketmq.tools.command.controller.GetControllerConfigSubCommand; import org.apache.rocketmq.tools.command.controller.GetControllerMetaDataSubCommand; import org.apache.rocketmq.tools.command.controller.UpdateControllerConfigSubCommand; +import org.apache.rocketmq.tools.command.controller.ReElectMasterSubCommand; import org.apache.rocketmq.tools.command.export.ExportConfigsCommand; import org.apache.rocketmq.tools.command.export.ExportMetadataCommand; import org.apache.rocketmq.tools.command.export.ExportMetricsCommand; @@ -264,6 +265,7 @@ public class MQAdminStartup { initCommand(new GetControllerConfigSubCommand()); initCommand(new UpdateControllerConfigSubCommand()); + initCommand(new ReElectMasterSubCommand()); } private static void initLogback() throws JoranException { diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/controller/ReElectMasterSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/controller/ReElectMasterSubCommand.java new file mode 100644 index 0000000000..47e8dfadc0 --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/controller/ReElectMasterSubCommand.java @@ -0,0 +1,92 @@ +/* + * 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.controller; + +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.BrokerMemberGroup; +import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader; +import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; +import org.apache.rocketmq.tools.command.SubCommand; +import org.apache.rocketmq.tools.command.SubCommandException; + +public class ReElectMasterSubCommand implements SubCommand { + + @Override + public String commandName() { + return "electMaster"; + } + + @Override + public String commandDesc() { + return "Re-elect the specified broker as master"; + } + + @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("b", "brokerAddress", true, "The address of the broker which requires to become master"); + opt.setRequired(true); + options.addOption(opt); + + opt = new Option("n", "brokerName", true, "The broker name of the replicas that require to be manipulated"); + opt.setRequired(true); + options.addOption(opt); + + opt = new Option("c", "clusterName", true, "the clusterName of broker"); + opt.setRequired(true); + 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())); + String controllerAddress = commandLine.getOptionValue("a").trim(); + String clusterName = commandLine.getOptionValue('c').trim(); + String brokerName = commandLine.getOptionValue('n').trim(); + String brokerAddress = commandLine.getOptionValue("b").trim(); + + try { + defaultMQAdminExt.start(); + final ElectMasterResponseHeader metaData = defaultMQAdminExt.electMaster(controllerAddress, clusterName, brokerName, brokerAddress); + System.out.printf("\n#ClusterName\t%s", clusterName); + System.out.printf("\n#BrokerName\t%s", brokerName); + System.out.printf("\n#BrokerMasterAddr\t%s", metaData.getNewMasterAddress()); + System.out.printf("\n#MasterEpoch\t%s", metaData.getMasterEpoch()); + System.out.printf("\n#SyncStateSetEpoch\t%s\n", metaData.getSyncStateSetEpoch()); + BrokerMemberGroup brokerMemberGroup = metaData.getBrokerMemberGroup(); + if (null != brokerMemberGroup && null != brokerMemberGroup.getBrokerAddrs()) { + brokerMemberGroup.getBrokerAddrs().forEach((key, value) -> System.out.printf("\t#Broker\t%d\t%s\n", key, value)); + } + } catch (Exception e) { + throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); + } finally { + defaultMQAdminExt.shutdown(); + } + + } +}