[ISSUE #4791] Add elect master command for admin CLI (#4798)

2026-09-21 05:44:03 +08:00 · 2022-08-11 10:06:46 +08:00
parent 9b87205c6d
commit 06ca14a508
10 changed files with 330 additions and 55 deletions
@@ -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<String, Properties> getControllerConfig(List<String> controllerServers) throws InterruptedException, RemotingTimeoutException,
+    public Map<String, Properties> getControllerConfig(
+        List<String> 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<String> 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);
+    }
 }
@@ -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<String> controllers) throws InterruptedException, RemotingConnectException, UnsupportedEncodingException,
        RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
        this.mqClientInstance.getMQClientAPIImpl().updateControllerConfig(properties, controllers, timeoutMillis);
@@ -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<TopicStatsTable> 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<String> 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<TopicList> 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<QueueTimeSpan> 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<String> brokers) throws RemotingException, InterruptedException, MQBrokerException;
+    InSyncStateData getInSyncStateData(String controllerAddress,
+        List<String> 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.
     * <br>
@@ -421,7 +432,8 @@ public interface MQAdminExt extends MQAdmin {
     *
     * @return The fetched controller config
     */
-    Map<String, Properties> getControllerConfig(List<String> controllerServers) throws InterruptedException, RemotingTimeoutException,
+    Map<String, Properties> getControllerConfig(
+        List<String> controllerServers) throws InterruptedException, RemotingTimeoutException,
        RemotingSendRequestException, RemotingConnectException, MQClientException, UnsupportedEncodingException;

    /**
@@ -433,4 +445,18 @@ public interface MQAdminExt extends MQAdmin {
        final List<String> 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;
 }
@@ -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 {
@@ -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();
+        }
+
+    }
+}