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 c8f5ccf38c74ec41bc9bea5c878de6e09490b71a..44a5113293641938efe4d48f360e44413f497c83 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 @@ -203,7 +203,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { } @Override - public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) { + public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) + throws InterruptedException, RemotingException, MQClientException, MQBrokerException { return defaultMQAdminExtImpl.examineSubscriptionGroupConfig(addr, group); } 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 c7636d907214b788bbd602b3c95ee8c7178b949a..55cca540f9b19d602ea6ed62af20cb65f5ba4df4 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 @@ -217,8 +217,10 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner { } @Override - public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) { - return null; + public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) + throws InterruptedException, RemotingException, MQClientException, MQBrokerException { + SubscriptionGroupWrapper wrapper = this.mqClientInstance.getMQClientAPIImpl().getAllSubscriptionGroup(addr, timeoutMillis); + return wrapper.getSubscriptionGroupTable().get(group); } @Override 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 a06c2c45e654181cb415fea1dc81df8806c728bc..d7010561a7ff0c6357ab8b8a62b5ea95cccfb247 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 @@ -90,7 +90,7 @@ public interface MQAdminExt extends MQAdmin { final SubscriptionGroupConfig config) throws RemotingException, MQBrokerException, InterruptedException, MQClientException; - SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, final String group); + SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, final String group) throws InterruptedException, RemotingException, MQClientException, MQBrokerException; TopicConfig examineTopicConfig(final String addr, final String topic); 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 25434dae4f85b2c4ec11a787b940c612351d203d..4411a6cc1d9819ea15d0a16ca6f84020ef90cf17 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 @@ -49,6 +49,7 @@ import org.apache.rocketmq.tools.command.connection.ProducerConnectionSubCommand import org.apache.rocketmq.tools.command.consumer.ConsumerProgressSubCommand; import org.apache.rocketmq.tools.command.consumer.ConsumerStatusSubCommand; import org.apache.rocketmq.tools.command.consumer.DeleteSubscriptionGroupCommand; +import org.apache.rocketmq.tools.command.consumer.GetConsumerConfigSubCommand; import org.apache.rocketmq.tools.command.consumer.StartMonitoringSubCommand; import org.apache.rocketmq.tools.command.consumer.UpdateSubGroupSubCommand; import org.apache.rocketmq.tools.command.message.CheckMsgSendRTCommand; @@ -204,6 +205,7 @@ public class MQAdminStartup { initCommand(new GetNamesrvConfigCommand()); initCommand(new UpdateNamesrvConfigCommand()); initCommand(new GetBrokerConfigCommand()); + initCommand(new GetConsumerConfigSubCommand()); initCommand(new QueryConsumeQueueCommand()); initCommand(new SendMessageCommand()); diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommand.java new file mode 100644 index 0000000000000000000000000000000000000000..be2b9466881b77976f33dc7912138ce53ba15cd4 --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommand.java @@ -0,0 +1,146 @@ +/* + * 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.consumer; + +import java.lang.reflect.Field; +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.commons.collections.CollectionUtils; +import org.apache.rocketmq.common.protocol.body.ClusterInfo; +import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig; +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 GetConsumerConfigSubCommand implements SubCommand { + + @Override + public String commandName() { + return "getConsumerConfig"; + } + + @Override + public String commandDesc() { + return "Get consumer config by subscription group name!"; + } + + @Override + public Options buildCommandlineOptions(final Options options) { + Option opt = new Option("g", "groupName", true, "subscription group name"); + opt.setRequired(true); + options.addOption(opt); + return options; + } + + @Override + public void execute(CommandLine commandLine, Options options, + RPCHook rpcHook) throws SubCommandException { + DefaultMQAdminExt adminExt = new DefaultMQAdminExt(rpcHook); + adminExt.setInstanceName(Long.toString(System.currentTimeMillis())); + String groupName = commandLine.getOptionValue('g').trim(); + try { + adminExt.start(); + List consumerConfigInfoList = new ArrayList<>(); + ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo(); + Map> clusterAddrTable = clusterInfo.getClusterAddrTable(); + for (String brokerName : clusterInfo.getBrokerAddrTable().keySet()) { + String clusterName = this.getClusterName(brokerName, clusterAddrTable); + String brokerAddress = clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(); + SubscriptionGroupConfig subscriptionGroupConfig = adminExt.examineSubscriptionGroupConfig(brokerAddress, groupName); + if (subscriptionGroupConfig == null) { + continue; + } + consumerConfigInfoList.add(new ConsumerConfigInfo(clusterName, brokerName, subscriptionGroupConfig)); + } + if (CollectionUtils.isEmpty(consumerConfigInfoList)) { + return; + } + for (ConsumerConfigInfo info : consumerConfigInfoList) { + System.out.printf("=============================%s:%s=============================\n", + info.getClusterName(), info.getBrokerName()); + SubscriptionGroupConfig config = info.getSubscriptionGroupConfig(); + Field[] fields = config.getClass().getDeclaredFields(); + for (Field field : fields) { + field.setAccessible(true); + if (field.get(config) != null) { + System.out.printf("%s%-40s= %s\n", "", field.getName(), field.get(config).toString()); + } else { + System.out.printf("%s%-40s= %s\n", "", field.getName(), ""); + } + } + } + } catch (Exception e) { + throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); + } finally { + adminExt.shutdown(); + } + } + + private String getClusterName(String brokeName, Map> clusterAddrTable) { + for (Map.Entry> entry : clusterAddrTable.entrySet()) { + Set brokerNameSet = entry.getValue(); + if (brokerNameSet.contains(brokeName)) { + return entry.getKey(); + } + } + return null; + } +} + +class ConsumerConfigInfo { + private String clusterName; + + private String brokerName; + + private SubscriptionGroupConfig subscriptionGroupConfig; + + public ConsumerConfigInfo(String clusterName, String brokerName, SubscriptionGroupConfig subscriptionGroupConfig) { + this.clusterName = clusterName; + this.brokerName = brokerName; + this.subscriptionGroupConfig = subscriptionGroupConfig; + } + + public String getClusterName() { + return clusterName; + } + + public void setClusterName(String clusterName) { + this.clusterName = clusterName; + } + + public String getBrokerName() { + return brokerName; + } + + public void setBrokerName(String brokerNameList) { + this.brokerName = brokerName; + } + + public SubscriptionGroupConfig getSubscriptionGroupConfig() { + return subscriptionGroupConfig; + } + + public void setSubscriptionGroupConfig(SubscriptionGroupConfig subscriptionGroupConfig) { + this.subscriptionGroupConfig = subscriptionGroupConfig; + } +} diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommandTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommandTest.java new file mode 100644 index 0000000000000000000000000000000000000000..1ec68ff02ba380e5ff2357fb4dd379d9a158214c --- /dev/null +++ b/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommandTest.java @@ -0,0 +1,83 @@ +/* + * 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.consumer; + +import java.io.UnsupportedEncodingException; +import java.lang.reflect.Field; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.PosixParser; +import org.apache.rocketmq.client.ClientConfig; +import org.apache.rocketmq.client.exception.MQBrokerException; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.impl.MQClientAPIImpl; +import org.apache.rocketmq.client.impl.MQClientManager; +import org.apache.rocketmq.client.impl.factory.MQClientInstance; +import org.apache.rocketmq.remoting.exception.RemotingConnectException; +import org.apache.rocketmq.remoting.exception.RemotingSendRequestException; +import org.apache.rocketmq.remoting.exception.RemotingTimeoutException; +import org.apache.rocketmq.srvutil.ServerUtil; +import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; +import org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl; +import org.apache.rocketmq.tools.command.SubCommandException; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Ignore; +import org.junit.Test; + +import static org.mockito.Mockito.mock; + +public class GetConsumerConfigSubCommandTest { + private static DefaultMQAdminExt defaultMQAdminExt; + private static DefaultMQAdminExtImpl defaultMQAdminExtImpl; + private static MQClientInstance mqClientInstance = MQClientManager.getInstance().getOrCreateMQClientInstance(new ClientConfig()); + private static MQClientAPIImpl mQClientAPIImpl; + + @BeforeClass + public static void init() throws NoSuchFieldException, IllegalAccessException, InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException, MQBrokerException, UnsupportedEncodingException { + mQClientAPIImpl = mock(MQClientAPIImpl.class); + defaultMQAdminExt = new DefaultMQAdminExt(); + defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000); + + Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance"); + field.setAccessible(true); + field.set(defaultMQAdminExtImpl, mqClientInstance); + field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl"); + field.setAccessible(true); + field.set(mqClientInstance, mQClientAPIImpl); + field = DefaultMQAdminExt.class.getDeclaredField("defaultMQAdminExtImpl"); + field.setAccessible(true); + field.set(defaultMQAdminExt, defaultMQAdminExtImpl); + } + + @AfterClass + public static void terminate() { + defaultMQAdminExt.shutdown(); + } + + @Ignore + @Test + public void testExecute() throws SubCommandException { + System.setProperty("rocketmq.namesrv.addr", "127.0.0.1:9876"); + GetConsumerConfigSubCommand cmd = new GetConsumerConfigSubCommand(); + Options options = ServerUtil.buildCommandlineOptions(new Options()); + String[] subargs = new String[] {"-g group_test"}; + final CommandLine commandLine = + ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs, cmd.buildCommandlineOptions(options), new PosixParser()); + cmd.execute(commandLine, options, null); + } +} \ No newline at end of file