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 14df64632b7..bddb47e70ec 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 @@ -91,6 +91,7 @@ import org.apache.rocketmq.tools.command.message.DumpCompactionLogCommand; import org.apache.rocketmq.tools.command.message.PrintMessageByQueueCommand; import org.apache.rocketmq.tools.command.message.PrintMessageSubCommand; +import org.apache.rocketmq.tools.command.message.PushConsumeMessageCommand; import org.apache.rocketmq.tools.command.message.QueryMsgByIdSubCommand; import org.apache.rocketmq.tools.command.message.QueryMsgByKeySubCommand; import org.apache.rocketmq.tools.command.message.QueryMsgByOffsetSubCommand; @@ -262,6 +263,7 @@ public static void initCommand() { initCommand(new QueryConsumeQueueCommand()); initCommand(new SendMessageCommand()); initCommand(new ConsumeMessageCommand()); + initCommand(new PushConsumeMessageCommand()); initCommand(new UpdateStaticTopicSubCommand()); initCommand(new RemappingStaticTopicSubCommand()); diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageCommand.java new file mode 100644 index 00000000000..78d6fc8b6dc --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageCommand.java @@ -0,0 +1,193 @@ +/* + * 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.PrintStream; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely; +import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.tools.command.SubCommand; +import org.apache.rocketmq.tools.command.SubCommandException; + +/** + * Starts a real push consumer from mqadmin so operators can inspect message delivery, progress, and tracing without + * compiling a standalone demo application. + */ +public class PushConsumeMessageCommand implements SubCommand { + private final ConsumerFactory consumerFactory; + private final PrintStream out; + + public PushConsumeMessageCommand() { + this(DefaultMQPushConsumer::new, System.out); + } + + PushConsumeMessageCommand(ConsumerFactory consumerFactory, PrintStream out) { + this.consumerFactory = consumerFactory; + this.out = out; + } + + @Override + public String commandName() { + return "consumeMessageByPush"; + } + + @Override + public String commandDesc() { + return "Consume and inspect messages with a temporary PushConsumer."; + } + + @Override + public Options buildCommandlineOptions(Options options) { + Option option = new Option("t", "topic", true, "Topic name"); + option.setRequired(true); + options.addOption(option); + + option = new Option("g", "consumerGroup", true, + "Consumer group used by the temporary PushConsumer"); + option.setRequired(true); + options.addOption(option); + + option = new Option("s", "subExpression", true, + "Subscription expression, for example '*' or 'TagA || TagB'. Default: *"); + option.setRequired(false); + options.addOption(option); + + option = new Option("c", "messageCount", true, + "Stop after receiving at least this many messages. 0 means unlimited. Default: 0"); + option.setRequired(false); + options.addOption(option); + + option = new Option("w", "maxWaitSeconds", true, + "Stop after this many seconds. 0 means unlimited. Default: 0"); + option.setRequired(false); + options.addOption(option); + + option = new Option("b", "batchSize", true, + "Maximum messages delivered to the listener per batch, from 1 to 1024. Default: 1"); + option.setRequired(false); + options.addOption(option); + + option = new Option("o", "orderly", true, + "Use orderly consumption. true or false. Default: false"); + option.setRequired(false); + options.addOption(option); + + option = new Option("m", "messageTraceEnabled", true, + "Enable message trace. true or false. Default: false"); + option.setRequired(false); + options.addOption(option); + + option = new Option("r", "customizedTraceTopic", true, + "Customized trace topic; requires messageTraceEnabled=true"); + option.setRequired(false); + options.addOption(option); + + option = new Option("d", "printBody", true, + "Print message body. true or false. Default: true"); + option.setRequired(false); + options.addOption(option); + + option = new Option("p", "printProperties", true, + "Print message properties. true or false. Default: false"); + option.setRequired(false); + options.addOption(option); + + option = new Option("a", "charset", true, "Charset used to decode message bodies. Default: UTF-8"); + option.setRequired(false); + options.addOption(option); + + option = new Option("i", "instanceName", true, + "Client instance name. Default: a unique mqadmin_push timestamp"); + option.setRequired(false); + options.addOption(option); + + option = new Option("f", "consumeFromWhere", true, + "Initial position for a new group: LAST, FIRST, or TIMESTAMP. Default: LAST"); + option.setRequired(false); + options.addOption(option); + + option = new Option("x", "consumeTimestamp", true, + "Start timestamp in yyyyMMddHHmmss format when consumeFromWhere=TIMESTAMP"); + option.setRequired(false); + options.addOption(option); + + return options; + } + + @Override + public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { + final PushConsumeMessageConfig config; + try { + config = PushConsumeMessageConfig.fromCommandLine(commandLine); + } catch (IllegalArgumentException e) { + throw new SubCommandException(commandName() + " command has invalid arguments: " + e.getMessage(), e); + } + + DefaultMQPushConsumer consumer = null; + PushConsumeMessagePrinter printer = new PushConsumeMessagePrinter(config, out); + PushConsumeMessageObserver observer = new PushConsumeMessageObserver(config, printer); + PushConsumeMessageObserver.CompletionReason completionReason = PushConsumeMessageObserver.CompletionReason.STOPPED; + try { + consumer = createConsumer(config, rpcHook); + configureConsumer(consumer, config, observer); + consumer.start(); + printer.printStarted(); + completionReason = observer.awaitCompletion(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + completionReason = PushConsumeMessageObserver.CompletionReason.INTERRUPTED; + } catch (Exception e) { + throw new SubCommandException(commandName() + " command failed", e); + } finally { + if (consumer != null) { + consumer.shutdown(); + } + printer.printSummary(observer.snapshot(), completionReason); + } + } + + private DefaultMQPushConsumer createConsumer(PushConsumeMessageConfig config, RPCHook rpcHook) { + return consumerFactory.create(config.getConsumerGroup(), rpcHook, new AllocateMessageQueueAveragely(), + config.isMessageTraceEnabled(), config.getCustomizedTraceTopic()); + } + + private void configureConsumer(DefaultMQPushConsumer consumer, PushConsumeMessageConfig config, + PushConsumeMessageObserver observer) throws Exception { + consumer.setInstanceName(config.getInstanceName()); + consumer.setConsumeMessageBatchMaxSize(config.getBatchSize()); + consumer.setConsumeFromWhere(config.getConsumeFromWhere()); + if (config.getConsumeTimestamp() != null) { + consumer.setConsumeTimestamp(config.getConsumeTimestamp()); + } + consumer.subscribe(config.getTopic(), config.getSubExpression()); + if (config.isOrderly()) { + consumer.registerMessageListener((org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly) observer); + } else { + consumer.registerMessageListener((org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently) observer); + } + } + + @FunctionalInterface + interface ConsumerFactory { + DefaultMQPushConsumer create(String consumerGroup, RPCHook rpcHook, + org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy allocateMessageQueueStrategy, + boolean enableMessageTrace, String customizedTraceTopic); + } +} diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageConfig.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageConfig.java new file mode 100644 index 00000000000..686bc11b129 --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageConfig.java @@ -0,0 +1,278 @@ +/* + * 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.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Locale; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; + +/** + * Validated configuration for the push-consumer diagnostic command. + */ +public class PushConsumeMessageConfig { + static final long DEFAULT_MAX_WAIT_SECONDS = 0; + static final long DEFAULT_MAX_MESSAGES = 0; + static final int DEFAULT_BATCH_SIZE = 1; + + private final String topic; + private final String consumerGroup; + private final String subExpression; + private final String instanceName; + private final long maxMessages; + private final long maxWaitMillis; + private final int batchSize; + private final boolean orderly; + private final boolean messageTraceEnabled; + private final String customizedTraceTopic; + private final boolean printBody; + private final boolean printProperties; + private final Charset charset; + private final ConsumeFromWhere consumeFromWhere; + private final String consumeTimestamp; + + PushConsumeMessageConfig(String topic, String consumerGroup, String subExpression, String instanceName, + long maxMessages, long maxWaitMillis, int batchSize, boolean orderly, boolean messageTraceEnabled, + String customizedTraceTopic, boolean printBody, boolean printProperties, Charset charset, + ConsumeFromWhere consumeFromWhere, String consumeTimestamp) { + this.topic = topic; + this.consumerGroup = consumerGroup; + this.subExpression = subExpression; + this.instanceName = instanceName; + this.maxMessages = maxMessages; + this.maxWaitMillis = maxWaitMillis; + this.batchSize = batchSize; + this.orderly = orderly; + this.messageTraceEnabled = messageTraceEnabled; + this.customizedTraceTopic = customizedTraceTopic; + this.printBody = printBody; + this.printProperties = printProperties; + this.charset = charset; + this.consumeFromWhere = consumeFromWhere; + this.consumeTimestamp = consumeTimestamp; + } + + public static PushConsumeMessageConfig fromCommandLine(CommandLine commandLine) { + String topic = requiredValue(commandLine, 't', "topic"); + String consumerGroup = requiredValue(commandLine, 'g', "consumerGroup"); + String subExpression = optionalValue(commandLine, 's', "*"); + String instanceName = optionalValue(commandLine, 'i', buildDefaultInstanceName()); + long maxMessages = parseNonNegativeLong(commandLine, 'c', DEFAULT_MAX_MESSAGES, "messageCount"); + long maxWaitSeconds = parseNonNegativeLong(commandLine, 'w', DEFAULT_MAX_WAIT_SECONDS, "maxWaitSeconds"); + int batchSize = parseBatchSize(commandLine); + boolean orderly = parseBoolean(commandLine, 'o', false, "orderly"); + boolean traceEnabled = parseBoolean(commandLine, 'm', false, "messageTraceEnabled"); + String traceTopic = optionalNullableValue(commandLine, 'r'); + boolean printBody = parseBoolean(commandLine, 'd', true, "printBody"); + boolean printProperties = parseBoolean(commandLine, 'p', false, "printProperties"); + Charset charset = parseCharset(commandLine); + ConsumeFromWhere consumeFromWhere = parseConsumeFromWhere(commandLine); + String consumeTimestamp = optionalNullableValue(commandLine, 'x'); + + if (consumeFromWhere == ConsumeFromWhere.CONSUME_FROM_TIMESTAMP && StringUtils.isBlank(consumeTimestamp)) { + throw new IllegalArgumentException("consumeTimestamp is required when consumeFromWhere is TIMESTAMP"); + } + if (consumeTimestamp != null && UtilAll.parseDate(consumeTimestamp, UtilAll.YYYYMMDDHHMMSS) == null) { + throw new IllegalArgumentException("consumeTimestamp must use yyyyMMddHHmmss format: " + consumeTimestamp); + } + if (consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_TIMESTAMP && consumeTimestamp != null) { + throw new IllegalArgumentException("consumeTimestamp can only be used with consumeFromWhere TIMESTAMP"); + } + if (!traceEnabled && traceTopic != null) { + throw new IllegalArgumentException("customizedTraceTopic requires messageTraceEnabled=true"); + } + + return new PushConsumeMessageConfig(topic, consumerGroup, subExpression, instanceName, maxMessages, + secondsToMillis(maxWaitSeconds), batchSize, orderly, traceEnabled, traceTopic, printBody, + printProperties, charset, consumeFromWhere, consumeTimestamp); + } + + private static long secondsToMillis(long seconds) { + try { + return Math.multiplyExact(seconds, 1000); + } catch (ArithmeticException e) { + throw new IllegalArgumentException("maxWaitSeconds is too large: " + seconds, e); + } + } + + private static String requiredValue(CommandLine commandLine, char option, String optionName) { + String value = optionalNullableValue(commandLine, option); + if (StringUtils.isBlank(value)) { + throw new IllegalArgumentException(optionName + " must not be blank"); + } + return value; + } + + private static String optionalValue(CommandLine commandLine, char option, String defaultValue) { + String value = optionalNullableValue(commandLine, option); + return value == null ? defaultValue : value; + } + + private static String optionalNullableValue(CommandLine commandLine, char option) { + if (!commandLine.hasOption(option)) { + return null; + } + String value = commandLine.getOptionValue(option); + if (value == null) { + return null; + } + value = value.trim(); + return value.isEmpty() ? null : value; + } + + private static long parseNonNegativeLong(CommandLine commandLine, char option, long defaultValue, + String optionName) { + String value = optionalNullableValue(commandLine, option); + if (value == null) { + return defaultValue; + } + + final long parsed; + try { + parsed = Long.parseLong(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException(optionName + " must be an integer: " + value, e); + } + if (parsed < 0) { + throw new IllegalArgumentException(optionName + " must be greater than or equal to zero"); + } + return parsed; + } + + private static int parseBatchSize(CommandLine commandLine) { + String value = optionalNullableValue(commandLine, 'b'); + if (value == null) { + return DEFAULT_BATCH_SIZE; + } + + final int parsed; + try { + parsed = Integer.parseInt(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("batchSize must be an integer: " + value, e); + } + if (parsed < 1 || parsed > 1024) { + throw new IllegalArgumentException("batchSize must be between 1 and 1024"); + } + return parsed; + } + + private static boolean parseBoolean(CommandLine commandLine, char option, boolean defaultValue, + String optionName) { + String value = optionalNullableValue(commandLine, option); + if (value == null) { + return defaultValue; + } + if (!"true".equalsIgnoreCase(value) && !"false".equalsIgnoreCase(value)) { + throw new IllegalArgumentException(optionName + " must be true or false: " + value); + } + return Boolean.parseBoolean(value); + } + + private static Charset parseCharset(CommandLine commandLine) { + String value = optionalNullableValue(commandLine, 'a'); + if (value == null) { + return StandardCharsets.UTF_8; + } + try { + return Charset.forName(value); + } catch (RuntimeException e) { + throw new IllegalArgumentException("Unsupported charset: " + value, e); + } + } + + private static ConsumeFromWhere parseConsumeFromWhere(CommandLine commandLine) { + String value = optionalValue(commandLine, 'f', "LAST").toUpperCase(Locale.ROOT); + switch (value) { + case "LAST": + return ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET; + case "FIRST": + return ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET; + case "TIMESTAMP": + return ConsumeFromWhere.CONSUME_FROM_TIMESTAMP; + default: + throw new IllegalArgumentException("consumeFromWhere must be LAST, FIRST, or TIMESTAMP: " + value); + } + } + + private static String buildDefaultInstanceName() { + return "mqadmin_push_" + System.currentTimeMillis(); + } + + public String getTopic() { + return topic; + } + + public String getConsumerGroup() { + return consumerGroup; + } + + public String getSubExpression() { + return subExpression; + } + + public String getInstanceName() { + return instanceName; + } + + public long getMaxMessages() { + return maxMessages; + } + + public long getMaxWaitMillis() { + return maxWaitMillis; + } + + public int getBatchSize() { + return batchSize; + } + + public boolean isOrderly() { + return orderly; + } + + public boolean isMessageTraceEnabled() { + return messageTraceEnabled; + } + + public String getCustomizedTraceTopic() { + return customizedTraceTopic; + } + + public boolean isPrintBody() { + return printBody; + } + + public boolean isPrintProperties() { + return printProperties; + } + + public Charset getCharset() { + return charset; + } + + public ConsumeFromWhere getConsumeFromWhere() { + return consumeFromWhere; + } + + public String getConsumeTimestamp() { + return consumeTimestamp; + } +} diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageObserver.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageObserver.java new file mode 100644 index 00000000000..4a81c77427e --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageObserver.java @@ -0,0 +1,227 @@ +/* + * 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.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; + +/** + * Records messages delivered by a temporary push consumer and exposes a stable summary for the command output. + */ +public class PushConsumeMessageObserver implements MessageListenerConcurrently, MessageListenerOrderly { + private final PushConsumeMessageConfig config; + private final PushConsumeMessagePrinter printer; + private final CountDownLatch completionLatch = new CountDownLatch(1); + private final AtomicLong messageCount = new AtomicLong(); + private final AtomicLong bodyBytes = new AtomicLong(); + private final AtomicLong firstReceiveTimestamp = new AtomicLong(); + private final AtomicLong lastReceiveTimestamp = new AtomicLong(); + private final ConcurrentMap queueStatsTable = new ConcurrentHashMap<>(); + + public PushConsumeMessageObserver(PushConsumeMessageConfig config, PushConsumeMessagePrinter printer) { + this.config = config; + this.printer = printer; + } + + @Override + public ConsumeConcurrentlyStatus consumeMessage(List messages, ConsumeConcurrentlyContext context) { + record(messages, context.getMessageQueue()); + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + + @Override + public ConsumeOrderlyStatus consumeMessage(List messages, ConsumeOrderlyContext context) { + record(messages, context.getMessageQueue()); + return ConsumeOrderlyStatus.SUCCESS; + } + + void record(List messages, MessageQueue deliveryQueue) { + if (messages == null || messages.isEmpty()) { + return; + } + + for (MessageExt message : messages) { + long receiveTimestamp = System.currentTimeMillis(); + firstReceiveTimestamp.compareAndSet(0, receiveTimestamp); + lastReceiveTimestamp.set(receiveTimestamp); + long currentCount = messageCount.incrementAndGet(); + if (message.getBody() != null) { + bodyBytes.addAndGet(message.getBody().length); + } + + MessageQueue messageQueue = resolveMessageQueue(message, deliveryQueue); + queueStatsTable.computeIfAbsent(messageQueue, ignored -> new QueueStats()) + .record(message.getQueueOffset(), message.getStoreTimestamp()); + printer.printMessage(currentCount, receiveTimestamp, messageQueue, message); + } + + if (config.getMaxMessages() > 0 && messageCount.get() >= config.getMaxMessages()) { + completionLatch.countDown(); + } + } + + public CompletionReason awaitCompletion() throws InterruptedException { + if (config.getMaxMessages() == 0 && config.getMaxWaitMillis() == 0) { + completionLatch.await(); + return CompletionReason.STOPPED; + } + + if (config.getMaxWaitMillis() == 0) { + completionLatch.await(); + return CompletionReason.MESSAGE_LIMIT; + } + + boolean completed = completionLatch.await(config.getMaxWaitMillis(), TimeUnit.MILLISECONDS); + return completed ? CompletionReason.MESSAGE_LIMIT : CompletionReason.TIME_LIMIT; + } + + public void stop() { + completionLatch.countDown(); + } + + public Snapshot snapshot() { + List queues = new ArrayList<>(); + for (Map.Entry entry : queueStatsTable.entrySet()) { + QueueStats stats = entry.getValue(); + queues.add(new QueueSnapshot(entry.getKey(), stats.count.get(), stats.firstOffset.get(), + stats.lastOffset.get(), stats.lastStoreTimestamp.get())); + } + queues.sort(Comparator + .comparing((QueueSnapshot value) -> value.getMessageQueue().getBrokerName(), + Comparator.nullsFirst(String::compareTo)) + .thenComparingInt(value -> value.getMessageQueue().getQueueId())); + return new Snapshot(messageCount.get(), bodyBytes.get(), firstReceiveTimestamp.get(), + lastReceiveTimestamp.get(), queues); + } + + private MessageQueue resolveMessageQueue(MessageExt message, MessageQueue deliveryQueue) { + if (deliveryQueue != null) { + return deliveryQueue; + } + return new MessageQueue(message.getTopic(), message.getBrokerName(), message.getQueueId()); + } + + public enum CompletionReason { + MESSAGE_LIMIT, + TIME_LIMIT, + INTERRUPTED, + STOPPED + } + + private static class QueueStats { + private final AtomicLong count = new AtomicLong(); + private final AtomicLong firstOffset = new AtomicLong(Long.MAX_VALUE); + private final AtomicLong lastOffset = new AtomicLong(Long.MIN_VALUE); + private final AtomicLong lastStoreTimestamp = new AtomicLong(); + + private void record(long queueOffset, long storeTimestamp) { + count.incrementAndGet(); + firstOffset.accumulateAndGet(queueOffset, Math::min); + lastOffset.accumulateAndGet(queueOffset, Math::max); + lastStoreTimestamp.accumulateAndGet(storeTimestamp, Math::max); + } + } + + public static class Snapshot { + private final long messageCount; + private final long bodyBytes; + private final long firstReceiveTimestamp; + private final long lastReceiveTimestamp; + private final List queueSnapshots; + + Snapshot(long messageCount, long bodyBytes, long firstReceiveTimestamp, long lastReceiveTimestamp, + List queueSnapshots) { + this.messageCount = messageCount; + this.bodyBytes = bodyBytes; + this.firstReceiveTimestamp = firstReceiveTimestamp; + this.lastReceiveTimestamp = lastReceiveTimestamp; + this.queueSnapshots = queueSnapshots; + } + + public long getMessageCount() { + return messageCount; + } + + public long getBodyBytes() { + return bodyBytes; + } + + public long getFirstReceiveTimestamp() { + return firstReceiveTimestamp; + } + + public long getLastReceiveTimestamp() { + return lastReceiveTimestamp; + } + + public List getQueueSnapshots() { + return queueSnapshots; + } + } + + public static class QueueSnapshot { + private final MessageQueue messageQueue; + private final long messageCount; + private final long firstOffset; + private final long lastOffset; + private final long lastStoreTimestamp; + + QueueSnapshot(MessageQueue messageQueue, long messageCount, long firstOffset, long lastOffset, + long lastStoreTimestamp) { + this.messageQueue = messageQueue; + this.messageCount = messageCount; + this.firstOffset = firstOffset; + this.lastOffset = lastOffset; + this.lastStoreTimestamp = lastStoreTimestamp; + } + + public MessageQueue getMessageQueue() { + return messageQueue; + } + + public long getMessageCount() { + return messageCount; + } + + public long getFirstOffset() { + return firstOffset; + } + + public long getLastOffset() { + return lastOffset; + } + + public long getLastStoreTimestamp() { + return lastStoreTimestamp; + } + } +} diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessagePrinter.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessagePrinter.java new file mode 100644 index 00000000000..b993886473d --- /dev/null +++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PushConsumeMessagePrinter.java @@ -0,0 +1,149 @@ +/* + * 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.PrintStream; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.Map; +import java.util.TreeMap; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.remoting.common.RemotingHelper; + +/** + * Human-readable output for the push-consumer diagnostic command. + */ +public class PushConsumeMessagePrinter { + private static final String DATE_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS"; + + private final PushConsumeMessageConfig config; + private final PrintStream out; + + public PushConsumeMessagePrinter(PushConsumeMessageConfig config, PrintStream out) { + this.config = config; + this.out = out; + } + + public void printStarted() { + out.printf("Push consumer started. topic=%s, group=%s, subscription=%s, instance=%s, orderly=%s, trace=%s%n", + config.getTopic(), config.getConsumerGroup(), config.getSubExpression(), config.getInstanceName(), + config.isOrderly(), config.isMessageTraceEnabled()); + if (config.getMaxMessages() == 0 && config.getMaxWaitMillis() == 0) { + out.println("The command will keep consuming until the process is interrupted."); + } else { + out.printf("Stop condition: messageCount=%s, maxWaitMillis=%s%n", + config.getMaxMessages() == 0 ? "unlimited" : Long.toString(config.getMaxMessages()), + config.getMaxWaitMillis() == 0 ? "unlimited" : Long.toString(config.getMaxWaitMillis())); + } + } + + public synchronized void printMessage(long sequence, long receiveTimestamp, MessageQueue messageQueue, + MessageExt message) { + byte[] body = message.getBody(); + int bodySize = body == null ? 0 : body.length; + out.printf("%n#%d receiveTime=%s broker=%s queueId=%d queueOffset=%d%n", + sequence, formatTimestamp(receiveTimestamp), messageQueue.getBrokerName(), messageQueue.getQueueId(), + message.getQueueOffset()); + out.printf("msgId=%s keys=%s tags=%s reconsumeTimes=%d bodySize=%d%n", + valueOrDash(message.getMsgId()), valueOrDash(message.getKeys()), valueOrDash(message.getTags()), + message.getReconsumeTimes(), bodySize); + out.printf("bornTime=%s storeTime=%s bornHost=%s storeHost=%s%n", + formatTimestamp(message.getBornTimestamp()), formatTimestamp(message.getStoreTimestamp()), + socketAddress(message.getBornHost()), socketAddress(message.getStoreHost())); + + if (config.isPrintProperties()) { + printProperties(message.getProperties()); + } + if (config.isPrintBody()) { + out.printf("body=%s%n", decodeBody(body)); + } + } + + public void printSummary(PushConsumeMessageObserver.Snapshot snapshot, + PushConsumeMessageObserver.CompletionReason completionReason) { + out.printf("%nPush consumer stopped. reason=%s, messages=%d, bodyBytes=%d, elapsedMillis=%d%n", + completionReason, snapshot.getMessageCount(), snapshot.getBodyBytes(), elapsedMillis(snapshot)); + if (snapshot.getQueueSnapshots().isEmpty()) { + out.println("No messages were received."); + return; + } + + out.printf("%-32s %-8s %-10s %-14s %-14s %s%n", "#Broker", "#Queue", "#Messages", "#FirstOffset", + "#LastOffset", "#LastStoreTime"); + for (PushConsumeMessageObserver.QueueSnapshot queue : snapshot.getQueueSnapshots()) { + MessageQueue messageQueue = queue.getMessageQueue(); + out.printf("%-32s %-8d %-10d %-14d %-14d %s%n", messageQueue.getBrokerName(), messageQueue.getQueueId(), + queue.getMessageCount(), queue.getFirstOffset(), queue.getLastOffset(), + formatTimestamp(queue.getLastStoreTimestamp())); + } + } + + private void printProperties(Map properties) { + if (properties == null || properties.isEmpty()) { + out.println("properties={}"); + return; + } + + Map visibleProperties = new TreeMap<>(properties); + visibleProperties.remove(MessageConst.PROPERTY_WAIT_STORE_MSG_OK); + out.printf("properties=%s%n", visibleProperties); + } + + private String decodeBody(byte[] body) { + if (body == null) { + return ""; + } + try { + CharBuffer decoded = config.getCharset().newDecoder() + .onMalformedInput(CodingErrorAction.REPLACE) + .onUnmappableCharacter(CodingErrorAction.REPLACE) + .decode(ByteBuffer.wrap(body)); + return decoded.toString(); + } catch (CharacterCodingException e) { + return ""; + } + } + + private long elapsedMillis(PushConsumeMessageObserver.Snapshot snapshot) { + if (snapshot.getFirstReceiveTimestamp() == 0 || snapshot.getLastReceiveTimestamp() == 0) { + return 0; + } + return Math.max(0, snapshot.getLastReceiveTimestamp() - snapshot.getFirstReceiveTimestamp()); + } + + private String formatTimestamp(long timestamp) { + if (timestamp <= 0) { + return "-"; + } + return new SimpleDateFormat(DATE_PATTERN).format(new Date(timestamp)); + } + + private String valueOrDash(Object value) { + return value == null || value.toString().isEmpty() ? "-" : value.toString(); + } + + private String socketAddress(SocketAddress address) { + return valueOrDash(address == null ? null : RemotingHelper.parseSocketAddressAddr(address)); + } +} diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageCommandTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageCommandTest.java new file mode 100644 index 00000000000..4e55a416ec6 --- /dev/null +++ b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageCommandTest.java @@ -0,0 +1,262 @@ +/* + * 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.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.DefaultParser; +import org.apache.commons.cli.Options; +import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.tools.command.SubCommandException; +import org.junit.Assert; +import org.junit.Test; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +public class PushConsumeMessageCommandTest { + + @Test + public void testCommandMetadataAndOptions() { + PushConsumeMessageCommand command = new PushConsumeMessageCommand(); + Options options = command.buildCommandlineOptions(new Options()); + + Assert.assertEquals("consumeMessageByPush", command.commandName()); + Assert.assertTrue(command.commandDesc().contains("PushConsumer")); + Assert.assertTrue(options.getOption("topic").isRequired()); + Assert.assertTrue(options.getOption("consumerGroup").isRequired()); + Assert.assertNotNull(options.getOption("messageCount")); + Assert.assertNotNull(options.getOption("maxWaitSeconds")); + Assert.assertNotNull(options.getOption("messageTraceEnabled")); + Assert.assertNotNull(options.getOption("consumeTimestamp")); + } + + @Test + public void testConcurrentConsumerConfigurationAndDelivery() throws Exception { + DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class); + AtomicReference listener = new AtomicReference<>(); + doAnswer(invocation -> { + listener.set(invocation.getArgument(0)); + return null; + }).when(consumer).registerMessageListener(any(MessageListenerConcurrently.class)); + doAnswer(invocation -> { + MessageQueue queue = new MessageQueue("TopicA", "broker-a", 2); + ConsumeConcurrentlyStatus status = listener.get().consumeMessage( + Collections.singletonList(message()), new ConsumeConcurrentlyContext(queue)); + Assert.assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status); + return null; + }).when(consumer).start(); + + FactoryCapture factory = new FactoryCapture(consumer); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + PushConsumeMessageCommand command = new PushConsumeMessageCommand(factory, new PrintStream(output)); + CommandLine commandLine = parse(command, + "-t", "TopicA", "-g", "GroupA", "-s", "TagA", "-c", "1", "-b", "16", + "-i", "diagnostic", "-f", "first", "-d", "true", "-p", "true"); + + command.execute(commandLine, new Options(), null); + + Assert.assertEquals(1, factory.calls.get()); + Assert.assertEquals("GroupA", factory.consumerGroup.get()); + Assert.assertFalse(factory.traceEnabled); + verify(consumer).setInstanceName("diagnostic"); + verify(consumer).setConsumeMessageBatchMaxSize(16); + verify(consumer).setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); + verify(consumer, never()).setConsumeTimestamp(anyString()); + verify(consumer).subscribe("TopicA", "TagA"); + verify(consumer).start(); + verify(consumer).shutdown(); + String text = new String(output.toByteArray(), StandardCharsets.UTF_8); + Assert.assertTrue(text.contains("Push consumer started")); + Assert.assertTrue(text.contains("body=payload")); + Assert.assertTrue(text.contains("reason=MESSAGE_LIMIT, messages=1")); + } + + @Test + public void testOrderlyConsumerAndTraceConfiguration() throws Exception { + DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class); + AtomicReference listener = new AtomicReference<>(); + doAnswer(invocation -> { + listener.set(invocation.getArgument(0)); + return null; + }).when(consumer).registerMessageListener(any(MessageListenerOrderly.class)); + doAnswer(invocation -> { + MessageQueue queue = new MessageQueue("TopicA", "broker-a", 2); + ConsumeOrderlyStatus status = listener.get().consumeMessage(Collections.singletonList(message()), + new ConsumeOrderlyContext(queue)); + Assert.assertEquals(ConsumeOrderlyStatus.SUCCESS, status); + return null; + }).when(consumer).start(); + + FactoryCapture factory = new FactoryCapture(consumer); + PushConsumeMessageCommand command = new PushConsumeMessageCommand(factory, + new PrintStream(new ByteArrayOutputStream())); + CommandLine commandLine = parse(command, "-t", "TopicA", "-g", "GroupA", "-c", "1", "-o", "true", + "-m", "true", "-r", "TraceTopic", "-f", "timestamp", "-x", "20260718120000"); + + command.execute(commandLine, new Options(), null); + + Assert.assertTrue(factory.traceEnabled); + Assert.assertEquals("TraceTopic", factory.traceTopic.get()); + verify(consumer).setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP); + verify(consumer).setConsumeTimestamp("20260718120000"); + verify(consumer).registerMessageListener(any(MessageListenerOrderly.class)); + verify(consumer, never()).registerMessageListener(any(MessageListenerConcurrently.class)); + verify(consumer).shutdown(); + } + + @Test + public void testInvalidArgumentsDoNotCreateConsumer() throws Exception { + FactoryCapture factory = new FactoryCapture(mock(DefaultMQPushConsumer.class)); + PushConsumeMessageCommand command = new PushConsumeMessageCommand(factory, + new PrintStream(new ByteArrayOutputStream())); + CommandLine commandLine = parse(command, "-t", "TopicA", "-g", "GroupA", "-b", "0"); + + SubCommandException exception = Assert.assertThrows(SubCommandException.class, + () -> command.execute(commandLine, new Options(), null)); + + Assert.assertTrue(exception.getMessage().contains("invalid arguments")); + Assert.assertEquals(0, factory.calls.get()); + } + + @Test + public void testStartFailureStillShutsDownAndPrintsSummary() throws Exception { + DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class); + doThrow(new IllegalStateException("start failed")).when(consumer).start(); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + PushConsumeMessageCommand command = new PushConsumeMessageCommand(new FactoryCapture(consumer), + new PrintStream(output)); + CommandLine commandLine = parse(command, "-t", "TopicA", "-g", "GroupA", "-c", "1"); + + SubCommandException exception = Assert.assertThrows(SubCommandException.class, + () -> command.execute(commandLine, new Options(), null)); + + Assert.assertTrue(exception.getMessage().contains("command failed")); + verify(consumer).shutdown(); + Assert.assertTrue(new String(output.toByteArray(), StandardCharsets.UTF_8) + .contains("No messages were received")); + } + + @Test + public void testInterruptedWaitRestoresInterruptFlag() throws Exception { + DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class); + PushConsumeMessageCommand command = new PushConsumeMessageCommand(new FactoryCapture(consumer), + new PrintStream(new ByteArrayOutputStream())); + CommandLine commandLine = parse(command, "-t", "TopicA", "-g", "GroupA"); + + Thread.currentThread().interrupt(); + try { + command.execute(commandLine, new Options(), null); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + verify(consumer).shutdown(); + } + + @Test + public void testRpcHookPassedToFactory() throws Exception { + DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class); + AtomicReference listener = new AtomicReference<>(); + doAnswer(invocation -> { + listener.set(invocation.getArgument(0)); + return null; + }).when(consumer).registerMessageListener(any(MessageListenerConcurrently.class)); + doAnswer(invocation -> { + listener.get().consumeMessage(Collections.singletonList(message()), + new ConsumeConcurrentlyContext(new MessageQueue("TopicA", "broker-a", 2))); + return null; + }).when(consumer).start(); + FactoryCapture factory = new FactoryCapture(consumer); + PushConsumeMessageCommand command = new PushConsumeMessageCommand(factory, + new PrintStream(new ByteArrayOutputStream())); + RPCHook rpcHook = mock(RPCHook.class); + + command.execute(parse(command, "-t", "TopicA", "-g", "GroupA", "-c", "1"), new Options(), rpcHook); + + Assert.assertSame(rpcHook, factory.rpcHook.get()); + } + + private CommandLine parse(PushConsumeMessageCommand command, String... arguments) throws Exception { + Options options = command.buildCommandlineOptions(new Options()); + return new DefaultParser().parse(options, arguments); + } + + private MessageExt message() { + MessageExt message = new MessageExt(); + message.setTopic("TopicA"); + message.setBrokerName("broker-a"); + message.setQueueId(2); + message.setQueueOffset(10); + message.setMsgId("message-id"); + message.setBody("payload".getBytes(StandardCharsets.UTF_8)); + message.setBornTimestamp(100); + message.setStoreTimestamp(200); + message.setBornHost(new InetSocketAddress("127.0.0.1", 1000)); + message.setStoreHost(new InetSocketAddress("127.0.0.1", 2000)); + return message; + } + + private static class FactoryCapture implements PushConsumeMessageCommand.ConsumerFactory { + private final DefaultMQPushConsumer consumer; + private final AtomicInteger calls = new AtomicInteger(); + private final AtomicReference consumerGroup = new AtomicReference<>(); + private final AtomicReference rpcHook = new AtomicReference<>(); + private final AtomicReference allocationStrategy = new AtomicReference<>(); + private final AtomicReference traceTopic = new AtomicReference<>(); + private boolean traceEnabled; + + private FactoryCapture(DefaultMQPushConsumer consumer) { + this.consumer = consumer; + } + + @Override + public DefaultMQPushConsumer create(String consumerGroup, RPCHook rpcHook, + AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean enableMessageTrace, + String customizedTraceTopic) { + calls.incrementAndGet(); + this.consumerGroup.set(consumerGroup); + this.rpcHook.set(rpcHook); + this.allocationStrategy.set(allocateMessageQueueStrategy); + this.traceEnabled = enableMessageTrace; + this.traceTopic.set(customizedTraceTopic); + return consumer; + } + } +} diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageConfigTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageConfigTest.java new file mode 100644 index 00000000000..3d2decd021e --- /dev/null +++ b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageConfigTest.java @@ -0,0 +1,195 @@ +/* + * 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.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.consumer.ConsumeFromWhere; +import org.junit.Assert; +import org.junit.Test; + +public class PushConsumeMessageConfigTest { + + @Test + public void testDefaultConfiguration() throws Exception { + PushConsumeMessageConfig config = parse("-t", "TopicA", "-g", "GroupA"); + + Assert.assertEquals("TopicA", config.getTopic()); + Assert.assertEquals("GroupA", config.getConsumerGroup()); + Assert.assertEquals("*", config.getSubExpression()); + Assert.assertTrue(config.getInstanceName().startsWith("mqadmin_push_")); + Assert.assertEquals(0, config.getMaxMessages()); + Assert.assertEquals(0, config.getMaxWaitMillis()); + Assert.assertEquals(1, config.getBatchSize()); + Assert.assertFalse(config.isOrderly()); + Assert.assertFalse(config.isMessageTraceEnabled()); + Assert.assertNull(config.getCustomizedTraceTopic()); + Assert.assertTrue(config.isPrintBody()); + Assert.assertFalse(config.isPrintProperties()); + Assert.assertEquals(StandardCharsets.UTF_8, config.getCharset()); + Assert.assertEquals(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET, config.getConsumeFromWhere()); + Assert.assertNull(config.getConsumeTimestamp()); + } + + @Test + public void testCompleteConfiguration() throws Exception { + PushConsumeMessageConfig config = parse( + "-t", "TopicA", + "-g", "GroupA", + "-s", "TagA || TagB", + "-i", "diagnostic-instance", + "-c", "23", + "-w", "17", + "-b", "32", + "-o", "true", + "-m", "true", + "-r", "TraceTopic", + "-d", "false", + "-p", "true", + "-a", "UTF-16", + "-f", "timestamp", + "-x", "20260718120000"); + + Assert.assertEquals("TagA || TagB", config.getSubExpression()); + Assert.assertEquals("diagnostic-instance", config.getInstanceName()); + Assert.assertEquals(23, config.getMaxMessages()); + Assert.assertEquals(17_000, config.getMaxWaitMillis()); + Assert.assertEquals(32, config.getBatchSize()); + Assert.assertTrue(config.isOrderly()); + Assert.assertTrue(config.isMessageTraceEnabled()); + Assert.assertEquals("TraceTopic", config.getCustomizedTraceTopic()); + Assert.assertFalse(config.isPrintBody()); + Assert.assertTrue(config.isPrintProperties()); + Assert.assertEquals("UTF-16", config.getCharset().name()); + Assert.assertEquals(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP, config.getConsumeFromWhere()); + Assert.assertEquals("20260718120000", config.getConsumeTimestamp()); + } + + @Test + public void testFirstOffsetConfiguration() throws Exception { + PushConsumeMessageConfig config = parse("-t", "TopicA", "-g", "GroupA", "-f", "FiRsT"); + Assert.assertEquals(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET, config.getConsumeFromWhere()); + } + + @Test + public void testTrimValues() throws Exception { + PushConsumeMessageConfig config = parse("-t", " TopicA ", "-g", " GroupA ", "-s", " TagA "); + Assert.assertEquals("TopicA", config.getTopic()); + Assert.assertEquals("GroupA", config.getConsumerGroup()); + Assert.assertEquals("TagA", config.getSubExpression()); + } + + @Test + public void testNegativeMessageCountRejected() { + assertInvalid("messageCount must be greater", "-t", "TopicA", "-g", "GroupA", "-c", "-1"); + } + + @Test + public void testInvalidMessageCountRejected() { + assertInvalid("messageCount must be an integer", "-t", "TopicA", "-g", "GroupA", "-c", "many"); + } + + @Test + public void testNegativeWaitRejected() { + assertInvalid("maxWaitSeconds must be greater", "-t", "TopicA", "-g", "GroupA", "-w", "-2"); + } + + @Test + public void testInvalidWaitRejected() { + assertInvalid("maxWaitSeconds must be an integer", "-t", "TopicA", "-g", "GroupA", "-w", "later"); + } + + @Test + public void testBatchSizeBelowRangeRejected() { + assertInvalid("batchSize must be between", "-t", "TopicA", "-g", "GroupA", "-b", "0"); + } + + @Test + public void testBatchSizeAboveRangeRejected() { + assertInvalid("batchSize must be between", "-t", "TopicA", "-g", "GroupA", "-b", "1025"); + } + + @Test + public void testInvalidBatchSizeRejected() { + assertInvalid("batchSize must be an integer", "-t", "TopicA", "-g", "GroupA", "-b", "large"); + } + + @Test + public void testInvalidBooleanRejected() { + assertInvalid("orderly must be true or false", "-t", "TopicA", "-g", "GroupA", "-o", "yes"); + } + + @Test + public void testUnsupportedCharsetRejected() { + assertInvalid("Unsupported charset", "-t", "TopicA", "-g", "GroupA", "-a", "not-a-charset"); + } + + @Test + public void testUnknownConsumeFromWhereRejected() { + assertInvalid("consumeFromWhere must be", "-t", "TopicA", "-g", "GroupA", "-f", "middle"); + } + + @Test + public void testTimestampRequiredForTimestampMode() { + assertInvalid("consumeTimestamp is required", "-t", "TopicA", "-g", "GroupA", "-f", "timestamp"); + } + + @Test + public void testTimestampRejectedForOtherModes() { + assertInvalid("consumeTimestamp can only", "-t", "TopicA", "-g", "GroupA", "-x", "20260718120000"); + } + + @Test + public void testInvalidTimestampFormatRejected() { + assertInvalid("consumeTimestamp must use", "-t", "TopicA", "-g", "GroupA", "-f", "timestamp", + "-x", "2026-07-18"); + } + + @Test + public void testWaitOverflowRejected() { + assertInvalid("maxWaitSeconds is too large", "-t", "TopicA", "-g", "GroupA", "-w", + Long.toString(Long.MAX_VALUE)); + } + + @Test + public void testTraceTopicRequiresTraceEnabled() { + assertInvalid("customizedTraceTopic requires", "-t", "TopicA", "-g", "GroupA", "-r", "TraceTopic"); + } + + @Test + public void testBlankTopicRejected() { + assertInvalid("topic must not be blank", "-t", " ", "-g", "GroupA"); + } + + @Test + public void testBlankConsumerGroupRejected() { + assertInvalid("consumerGroup must not be blank", "-t", "TopicA", "-g", " "); + } + + private PushConsumeMessageConfig parse(String... arguments) throws Exception { + Options options = new PushConsumeMessageCommand().buildCommandlineOptions(new Options()); + CommandLine commandLine = new DefaultParser().parse(options, arguments); + return PushConsumeMessageConfig.fromCommandLine(commandLine); + } + + private void assertInvalid(String expectedMessage, String... arguments) { + IllegalArgumentException exception = Assert.assertThrows(IllegalArgumentException.class, () -> parse(arguments)); + Assert.assertTrue(exception.getMessage(), exception.getMessage().contains(expectedMessage)); + } +} diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageObserverTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageObserverTest.java new file mode 100644 index 00000000000..1b8cfc9e86c --- /dev/null +++ b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessageObserverTest.java @@ -0,0 +1,217 @@ +/* + * 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.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.junit.Assert; +import org.junit.Test; + +public class PushConsumeMessageObserverTest { + + @Test + public void testConcurrentMessagesRecordedByQueue() throws Exception { + TestContext context = createContext(2, 1_000); + MessageQueue queue = new MessageQueue("TopicA", "broker-a", 1); + MessageExt first = message("TopicA", "broker-a", 1, 10, "one"); + MessageExt second = message("TopicA", "broker-a", 1, 11, "second"); + + ConsumeConcurrentlyStatus status = context.observer.consumeMessage(Arrays.asList(first, second), + new ConsumeConcurrentlyContext(queue)); + + Assert.assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status); + Assert.assertEquals(PushConsumeMessageObserver.CompletionReason.MESSAGE_LIMIT, + context.observer.awaitCompletion()); + PushConsumeMessageObserver.Snapshot snapshot = context.observer.snapshot(); + Assert.assertEquals(2, snapshot.getMessageCount()); + Assert.assertEquals(9, snapshot.getBodyBytes()); + Assert.assertEquals(1, snapshot.getQueueSnapshots().size()); + Assert.assertEquals(2, snapshot.getQueueSnapshots().get(0).getMessageCount()); + Assert.assertEquals(10, snapshot.getQueueSnapshots().get(0).getFirstOffset()); + Assert.assertEquals(11, snapshot.getQueueSnapshots().get(0).getLastOffset()); + Assert.assertTrue(snapshot.getFirstReceiveTimestamp() > 0); + Assert.assertTrue(snapshot.getLastReceiveTimestamp() >= snapshot.getFirstReceiveTimestamp()); + } + + @Test + public void testOrderlyMessageReturnsSuccess() throws Exception { + TestContext context = createContext(1, 1_000); + MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0); + + ConsumeOrderlyStatus status = context.observer.consumeMessage( + Collections.singletonList(message("TopicA", "broker-a", 0, 3, "body")), + new ConsumeOrderlyContext(queue)); + + Assert.assertEquals(ConsumeOrderlyStatus.SUCCESS, status); + Assert.assertEquals(PushConsumeMessageObserver.CompletionReason.MESSAGE_LIMIT, + context.observer.awaitCompletion()); + } + + @Test + public void testEmptyBatchDoesNotChangeSnapshot() { + TestContext context = createContext(1, 1_000); + context.observer.record(Collections.emptyList(), new MessageQueue("TopicA", "broker-a", 0)); + + PushConsumeMessageObserver.Snapshot snapshot = context.observer.snapshot(); + Assert.assertEquals(0, snapshot.getMessageCount()); + Assert.assertEquals(0, snapshot.getBodyBytes()); + Assert.assertTrue(snapshot.getQueueSnapshots().isEmpty()); + } + + @Test + public void testNullBatchDoesNotChangeSnapshot() { + TestContext context = createContext(1, 1_000); + context.observer.record(null, new MessageQueue("TopicA", "broker-a", 0)); + Assert.assertEquals(0, context.observer.snapshot().getMessageCount()); + } + + @Test + public void testNullBodyDoesNotIncrementBytes() { + TestContext context = createContext(1, 1_000); + MessageExt message = message("TopicA", "broker-a", 0, 4, "body"); + message.setBody(null); + + context.observer.record(Collections.singletonList(message), null); + + Assert.assertEquals(1, context.observer.snapshot().getMessageCount()); + Assert.assertEquals(0, context.observer.snapshot().getBodyBytes()); + } + + @Test + public void testDeliveryQueuePreferredOverMessageMetadata() { + TestContext context = createContext(1, 1_000); + MessageExt message = message("TopicA", "message-broker", 7, 4, "body"); + MessageQueue deliveryQueue = new MessageQueue("TopicA", "delivery-broker", 2); + + context.observer.record(Collections.singletonList(message), deliveryQueue); + + MessageQueue recordedQueue = context.observer.snapshot().getQueueSnapshots().get(0).getMessageQueue(); + Assert.assertEquals("delivery-broker", recordedQueue.getBrokerName()); + Assert.assertEquals(2, recordedQueue.getQueueId()); + } + + @Test + public void testQueueMetadataFallsBackToMessage() { + TestContext context = createContext(1, 1_000); + context.observer.record(Collections.singletonList(message("TopicA", "broker-a", 3, 8, "body")), null); + + MessageQueue recordedQueue = context.observer.snapshot().getQueueSnapshots().get(0).getMessageQueue(); + Assert.assertEquals("TopicA", recordedQueue.getTopic()); + Assert.assertEquals("broker-a", recordedQueue.getBrokerName()); + Assert.assertEquals(3, recordedQueue.getQueueId()); + } + + @Test + public void testQueueSnapshotsAreSorted() { + TestContext context = createContext(3, 1_000); + context.observer.record(Collections.singletonList(message("TopicA", "broker-b", 1, 1, "a")), null); + context.observer.record(Collections.singletonList(message("TopicA", "broker-a", 2, 1, "b")), null); + context.observer.record(Collections.singletonList(message("TopicA", "broker-a", 0, 1, "c")), null); + + Assert.assertEquals("broker-a", context.observer.snapshot().getQueueSnapshots().get(0) + .getMessageQueue().getBrokerName()); + Assert.assertEquals(0, context.observer.snapshot().getQueueSnapshots().get(0) + .getMessageQueue().getQueueId()); + Assert.assertEquals(2, context.observer.snapshot().getQueueSnapshots().get(1) + .getMessageQueue().getQueueId()); + Assert.assertEquals("broker-b", context.observer.snapshot().getQueueSnapshots().get(2) + .getMessageQueue().getBrokerName()); + } + + @Test + public void testTimeLimitCompletion() throws Exception { + TestContext context = createContext(5, 5); + Assert.assertEquals(PushConsumeMessageObserver.CompletionReason.TIME_LIMIT, + context.observer.awaitCompletion()); + } + + @Test + public void testStopUnblocksUnlimitedObserver() throws Exception { + TestContext context = createContext(0, 0); + CompletableFuture result = CompletableFuture.supplyAsync(() -> { + try { + return context.observer.awaitCompletion(); + } catch (InterruptedException e) { + throw new AssertionError(e); + } + }); + + context.observer.stop(); + + Assert.assertEquals(PushConsumeMessageObserver.CompletionReason.STOPPED, + result.get(1, TimeUnit.SECONDS)); + } + + @Test + public void testMessageLimitCanBeExceededByBatch() throws Exception { + TestContext context = createContext(1, 1_000); + context.observer.record(Arrays.asList( + message("TopicA", "broker-a", 0, 1, "a"), + message("TopicA", "broker-a", 0, 2, "b")), null); + + Assert.assertEquals(PushConsumeMessageObserver.CompletionReason.MESSAGE_LIMIT, + context.observer.awaitCompletion()); + Assert.assertEquals(2, context.observer.snapshot().getMessageCount()); + } + + private TestContext createContext(long maxMessages, long maxWaitMillis) { + PushConsumeMessageConfig config = new PushConsumeMessageConfig("TopicA", "GroupA", "*", "instance", + maxMessages, maxWaitMillis, 1, false, false, null, true, false, StandardCharsets.UTF_8, + ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET, null); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + PushConsumeMessagePrinter printer = new PushConsumeMessagePrinter(config, new PrintStream(output)); + return new TestContext(new PushConsumeMessageObserver(config, printer), output); + } + + private MessageExt message(String topic, String brokerName, int queueId, long queueOffset, String body) { + MessageExt message = new MessageExt(); + message.setTopic(topic); + message.setBrokerName(brokerName); + message.setQueueId(queueId); + message.setQueueOffset(queueOffset); + message.setMsgId("msg-" + queueOffset); + message.setBody(body.getBytes(StandardCharsets.UTF_8)); + message.setBornTimestamp(100); + message.setStoreTimestamp(200 + queueOffset); + message.setBornHost(new InetSocketAddress("127.0.0.1", 1000)); + message.setStoreHost(new InetSocketAddress("127.0.0.1", 2000)); + return message; + } + + private static class TestContext { + private final PushConsumeMessageObserver observer; + private final ByteArrayOutputStream output; + + private TestContext(PushConsumeMessageObserver observer, ByteArrayOutputStream output) { + this.observer = observer; + this.output = output; + } + } +} diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessagePrinterTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessagePrinterTest.java new file mode 100644 index 00000000000..ca7c2c4a3a3 --- /dev/null +++ b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PushConsumeMessagePrinterTest.java @@ -0,0 +1,166 @@ +/* + * 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.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.junit.Assert; +import org.junit.Test; + +public class PushConsumeMessagePrinterTest { + + @Test + public void testMessageDetailsAndBodyPrinted() { + TestPrinter testPrinter = createPrinter(true, true, 5, 3_000); + MessageExt message = message(); + message.putUserProperty("z-user", "last"); + message.putUserProperty("a-user", "first"); + + testPrinter.printer.printMessage(1, 300, new MessageQueue("TopicA", "broker-a", 2), message); + + String output = testPrinter.output(); + Assert.assertTrue(output.contains("#1 receiveTime=")); + Assert.assertTrue(output.contains("broker=broker-a queueId=2 queueOffset=12")); + Assert.assertTrue(output.contains("msgId=message-id keys=key-a tags=TagA reconsumeTimes=3 bodySize=5")); + Assert.assertTrue(output.contains("bornHost=127.0.0.1:1000")); + Assert.assertTrue(output.contains("storeHost=127.0.0.1:2000")); + Assert.assertTrue(output.contains("a-user=first")); + Assert.assertTrue(output.contains("z-user=last")); + Assert.assertTrue(output.contains("body=hello")); + } + + @Test + public void testBodyAndPropertiesCanBeHidden() { + TestPrinter testPrinter = createPrinter(false, false, 1, 1_000); + testPrinter.printer.printMessage(1, 300, new MessageQueue("TopicA", "broker-a", 2), message()); + Assert.assertFalse(testPrinter.output().contains("body=")); + Assert.assertFalse(testPrinter.output().contains("properties=")); + } + + @Test + public void testNullBodyPrinted() { + TestPrinter testPrinter = createPrinter(true, false, 1, 1_000); + MessageExt message = message(); + message.setBody(null); + testPrinter.printer.printMessage(1, 300, new MessageQueue("TopicA", "broker-a", 2), message); + Assert.assertTrue(testPrinter.output().contains("body=")); + } + + @Test + public void testEmptyPropertiesPrinted() { + TestPrinter testPrinter = createPrinter(false, true, 1, 1_000); + MessageExt message = new MessageExt(); + message.setMsgId("message-id"); + message.setQueueOffset(12); + message.setBornHost(new InetSocketAddress("127.0.0.1", 1000)); + message.setStoreHost(new InetSocketAddress("127.0.0.1", 2000)); + testPrinter.printer.printMessage(1, 300, new MessageQueue("TopicA", "broker-a", 2), message); + Assert.assertTrue(testPrinter.output().contains("properties={}")); + } + + @Test + public void testStartedOutputForBoundedRun() { + TestPrinter testPrinter = createPrinter(true, false, 5, 3_000); + testPrinter.printer.printStarted(); + + Assert.assertTrue(testPrinter.output().contains("Push consumer started. topic=TopicA, group=GroupA")); + Assert.assertTrue(testPrinter.output().contains("Stop condition: messageCount=5, maxWaitMillis=3000")); + } + + @Test + public void testStartedOutputForUnlimitedRun() { + TestPrinter testPrinter = createPrinter(true, false, 0, 0); + testPrinter.printer.printStarted(); + Assert.assertTrue(testPrinter.output().contains("keep consuming until the process is interrupted")); + } + + @Test + public void testEmptySummary() { + TestPrinter testPrinter = createPrinter(true, false, 1, 1_000); + PushConsumeMessageObserver.Snapshot snapshot = new PushConsumeMessageObserver.Snapshot(0, 0, 0, 0, + Collections.emptyList()); + + testPrinter.printer.printSummary(snapshot, PushConsumeMessageObserver.CompletionReason.TIME_LIMIT); + + Assert.assertTrue(testPrinter.output().contains("reason=TIME_LIMIT, messages=0, bodyBytes=0")); + Assert.assertTrue(testPrinter.output().contains("No messages were received")); + } + + @Test + public void testQueueSummary() { + TestPrinter testPrinter = createPrinter(true, false, 1, 1_000); + PushConsumeMessageObserver.QueueSnapshot queue = new PushConsumeMessageObserver.QueueSnapshot( + new MessageQueue("TopicA", "broker-a", 2), 4, 10, 13, 500); + PushConsumeMessageObserver.Snapshot snapshot = new PushConsumeMessageObserver.Snapshot(4, 20, 100, 250, + Collections.singletonList(queue)); + + testPrinter.printer.printSummary(snapshot, PushConsumeMessageObserver.CompletionReason.MESSAGE_LIMIT); + + String output = testPrinter.output(); + Assert.assertTrue(output.contains("reason=MESSAGE_LIMIT, messages=4, bodyBytes=20, elapsedMillis=150")); + Assert.assertTrue(output.contains("broker-a")); + Assert.assertTrue(output.contains("10")); + Assert.assertTrue(output.contains("13")); + } + + private TestPrinter createPrinter(boolean printBody, boolean printProperties, long maxMessages, + long maxWaitMillis) { + PushConsumeMessageConfig config = new PushConsumeMessageConfig("TopicA", "GroupA", "*", "instance", + maxMessages, maxWaitMillis, 1, false, false, null, printBody, printProperties, StandardCharsets.UTF_8, + ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET, null); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + return new TestPrinter(new PushConsumeMessagePrinter(config, new PrintStream(output)), output); + } + + private MessageExt message() { + MessageExt message = new MessageExt(); + message.setTopic("TopicA"); + message.setBrokerName("broker-a"); + message.setQueueId(2); + message.setQueueOffset(12); + message.setMsgId("message-id"); + message.setKeys("key-a"); + message.setTags("TagA"); + message.setReconsumeTimes(3); + message.setBody("hello".getBytes(StandardCharsets.UTF_8)); + message.setBornTimestamp(100); + message.setStoreTimestamp(200); + message.setBornHost(new InetSocketAddress("127.0.0.1", 1000)); + message.setStoreHost(new InetSocketAddress("127.0.0.1", 2000)); + return message; + } + + private static class TestPrinter { + private final PushConsumeMessagePrinter printer; + private final ByteArrayOutputStream output; + + private TestPrinter(PushConsumeMessagePrinter printer, ByteArrayOutputStream output) { + this.printer = printer; + this.output = output; + } + + private String output() { + return new String(output.toByteArray(), StandardCharsets.UTF_8); + } + } +}