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 01a6fe36468..04614ed8f8a 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 @@ -88,6 +88,7 @@ import org.apache.rocketmq.tools.command.lite.TriggerLiteDispatchSubCommand; import org.apache.rocketmq.tools.command.message.CheckMsgSendRTCommand; import org.apache.rocketmq.tools.command.message.ConsumeMessageCommand; +import org.apache.rocketmq.tools.command.message.DecodeMessageIdCommand; import org.apache.rocketmq.tools.command.message.DumpCompactionLogCommand; import org.apache.rocketmq.tools.command.message.PrintMessageByQueueCommand; import org.apache.rocketmq.tools.command.message.PrintMessageSubCommand; @@ -262,6 +263,7 @@ public static void initCommand() { initCommand(new QueryConsumeQueueCommand()); initCommand(new SendMessageCommand()); initCommand(new ConsumeMessageCommand()); + initCommand(new DecodeMessageIdCommand()); initCommand(new UpdateStaticTopicSubCommand()); initCommand(new RemappingStaticTopicSubCommand()); diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommand.java new file mode 100644 index 00000000000..dc8b1a2955a --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommand.java @@ -0,0 +1,86 @@ +/* + * 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.message; + +import java.net.InetSocketAddress; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageId; +import org.apache.rocketmq.common.utils.NetworkUtil; +import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.tools.command.SubCommand; +import org.apache.rocketmq.tools.command.SubCommandException; + +public class DecodeMessageIdCommand implements SubCommand { + private static final String MESSAGE_ID_TYPE_UNIQUE = "unique"; + private static final String MESSAGE_ID_TYPE_OFFSET = "offset"; + + @Override + public String commandName() { + return "decodeMessageId"; + } + + @Override + public String commandDesc() { + return "Decode unique or offset message ID."; + } + + @Override + public Options buildCommandlineOptions(Options options) { + Option opt = new Option("i", "messageId", true, "message ID"); + opt.setRequired(true); + options.addOption(opt); + + opt = new Option("t", "messageIdType", true, "message ID type, support unique or offset. Default: unique"); + opt.setRequired(false); + options.addOption(opt); + return options; + } + + @Override + public void execute(final CommandLine commandLine, final Options options, + RPCHook rpcHook) throws SubCommandException { + String messageId = commandLine.getOptionValue('i').trim(); + String messageIdType = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : MESSAGE_ID_TYPE_UNIQUE; + + try { + if (MESSAGE_ID_TYPE_UNIQUE.equalsIgnoreCase(messageIdType)) { + System.out.printf("IP: %s%n", MessageClientIDSetter.getIPStrFromID(messageId)); + System.out.printf("Date: %s%n", UtilAll.formatDate(MessageClientIDSetter.getNearlyTimeFromID(messageId), UtilAll.YYYY_MM_DD_HH_MM_SS_SSS)); + } else if (MESSAGE_ID_TYPE_OFFSET.equalsIgnoreCase(messageIdType)) { + MessageId decodedMessageId = MessageDecoder.decodeMessageId(messageId); + System.out.printf("StoreHost: %s%n", formatStoreHost(decodedMessageId)); + System.out.printf("CommitLogOffset: %d%n", decodedMessageId.getOffset()); + } else { + throw new SubCommandException("Unsupported messageIdType: " + messageIdType); + } + } catch (SubCommandException e) { + throw e; + } catch (Exception e) { + throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); + } + } + + private String formatStoreHost(MessageId messageId) { + InetSocketAddress address = (InetSocketAddress) messageId.getAddress(); + return NetworkUtil.normalizeHostAddress(address.getAddress()) + ":" + address.getPort(); + } +} diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java deleted file mode 100644 index 532508c0b26..00000000000 --- a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.message; - -import org.apache.commons.cli.CommandLine; -import org.apache.commons.cli.Option; -import org.apache.commons.cli.Options; -import org.apache.rocketmq.common.UtilAll; -import org.apache.rocketmq.common.message.MessageClientIDSetter; -import org.apache.rocketmq.remoting.RPCHook; -import org.apache.rocketmq.tools.command.SubCommand; -import org.apache.rocketmq.tools.command.SubCommandException; - -public class DecodeMessageIdCommond implements SubCommand { - @Override - public String commandName() { - return "DecodeMessageId"; - } - - @Override - public String commandDesc() { - return "decode unique message ID"; - } - - @Override - public Options buildCommandlineOptions(Options options) { - Option opt = new Option("i", "messageId", true, "unique message ID"); - opt.setRequired(false); - options.addOption(opt); - return options; - } - - @Override - public void execute(final CommandLine commandLine, final Options options, - RPCHook rpcHook) throws SubCommandException { - String messageId = commandLine.getOptionValue('i').trim(); - - try { - System.out.printf("ip=%s", MessageClientIDSetter.getIPStrFromID(messageId)); - } catch (Exception e) { - e.printStackTrace(); - } - - try { - String date = UtilAll.formatDate(MessageClientIDSetter.getNearlyTimeFromID(messageId), UtilAll.YYYY_MM_DD_HH_MM_SS_SSS); - System.out.printf("date=%s", date); - } catch (Exception e) { - throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); - } - } -} diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommandTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommandTest.java new file mode 100644 index 00000000000..51fd2f05e68 --- /dev/null +++ b/tools/src/test/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommandTest.java @@ -0,0 +1,112 @@ +/* + * 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.message; + +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.DefaultParser; +import org.apache.commons.cli.Options; +import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageId; +import org.apache.rocketmq.common.utils.NetworkUtil; +import org.apache.rocketmq.srvutil.ServerUtil; +import org.apache.rocketmq.tools.command.SubCommandException; +import org.junit.Assert; +import org.junit.Test; + +public class DecodeMessageIdCommandTest { + @Test + public void testExecuteDecodeUniqueMessageIdByDefault() throws Exception { + String uniqueMessageId = MessageClientIDSetter.createUniqID(); + Thread.sleep(5L); + String output = execute("-i", uniqueMessageId); + + Assert.assertEquals(String.format("IP: %s%nDate: %s%n", + MessageClientIDSetter.getIPStrFromID(uniqueMessageId), + UtilAll.formatDate(MessageClientIDSetter.getNearlyTimeFromID(uniqueMessageId), UtilAll.YYYY_MM_DD_HH_MM_SS_SSS)), + output); + } + + @Test + public void testExecuteDecodeDocumentedOffsetMessageId() throws SubCommandException { + String output = execute("-i", "0A42333A00002A9F000000000134F1F5", "-t", "offset"); + + Assert.assertEquals(String.format("StoreHost: 10.66.51.58:10911%nCommitLogOffset: 20247029%n"), output); + } + + @Test + public void testExecuteDecodeIPv4OffsetMessageId() throws Exception { + long commitLogOffset = 123456L; + String offsetMessageId = MessageDecoder.createMessageId(new InetSocketAddress("127.0.0.1", 10911), commitLogOffset); + MessageId decodedMessageId = MessageDecoder.decodeMessageId(offsetMessageId); + + String output = execute("-i", offsetMessageId, "-t", "offset"); + + Assert.assertEquals(String.format("StoreHost: %s%nCommitLogOffset: %d%n", + formatStoreHost(decodedMessageId), commitLogOffset), output); + } + + @Test + public void testExecuteDecodeIPv6OffsetMessageId() throws Exception { + long commitLogOffset = 654321L; + String offsetMessageId = MessageDecoder.createMessageId( + new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 10912), commitLogOffset); + Assert.assertEquals(56, offsetMessageId.length()); + + String output = execute("-i", offsetMessageId, "-t", "offset"); + + Assert.assertEquals(String.format("StoreHost: [1050:0:0:0:5:600:300c:326b]:10912%nCommitLogOffset: 654321%n"), output); + } + + @Test + public void testExecuteWithUnsupportedMessageIdType() throws SubCommandException { + try { + execute("-i", MessageClientIDSetter.createUniqID(), "-t", "unknown"); + Assert.fail("Should fail when messageIdType is unsupported"); + } catch (SubCommandException e) { + Assert.assertTrue(e.getMessage().contains("Unsupported messageIdType: unknown")); + } + } + + private String execute(String... subargs) throws SubCommandException { + DecodeMessageIdCommand cmd = new DecodeMessageIdCommand(); + Options options = ServerUtil.buildCommandlineOptions(new Options()); + CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs, + cmd.buildCommandlineOptions(options), new DefaultParser()); + PrintStream out = System.out; + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + System.setOut(new PrintStream(bos)); + try { + cmd.execute(commandLine, options, null); + return new String(bos.toByteArray(), StandardCharsets.UTF_8); + } finally { + System.setOut(out); + } + } + + private String formatStoreHost(MessageId messageId) { + InetSocketAddress address = (InetSocketAddress) messageId.getAddress(); + return NetworkUtil.normalizeHostAddress(address.getAddress()) + ":" + address.getPort(); + } +}