#491 增加消息id被谁消费api，扩展id查询命令

Author: vintagewang

rocketmq-example/src/main/java/com/alibaba/rocketmq/example/simple/PushConsumer.java

@@ -37,8 +37,8 @@ public static void main(String[] args) throws InterruptedException, MQClientExce
          * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例<br>
          * 注意：ConsumerGroupName需要由应用来保证唯一
          */
-        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_222");
-        // consumer.setNamesrvAddr("10.235.170.7:9877");
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_001");
+        // consumer.setNamesrvAddr("10.235.169.72:9876");
         // consumer.setNamesrvAddr("127.0.0.1:9876");
 
         /**
@@ -66,8 +66,7 @@ public static void main(String[] args) throws InterruptedException, MQClientExce
             @Override
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                     ConsumeConcurrentlyContext context) {
-                // System.out.println(Thread.currentThread().getName() +
-                // " Receive New Messages: " + msgs);
+                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
 
                 MessageExt msg = msgs.get(0);
                 if (msg.getTopic().equals("TopicTest1")) {

rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

@@ -748,10 +748,12 @@ public boolean consumed(final MessageExt msg, final String group) throws Remotin
             MessageQueue mq = next.getKey();
             if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
                 BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
-                String addr = brokerData.getBrokerAddrs().get(0);
-                if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
-                    if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
-                        return true;
+                if (brokerData != null) {
+                    String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
+                    if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
+                        if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
+                            return true;
+                        }
                     }
                 }
             }
@@ -791,7 +793,9 @@ public List<MessageTrack> messageTrackDetail(MessageExt msg) throws RemotingExce
                             Entry<String, SubscriptionData> next = it.next();
                             if (next.getKey().equals(msg.getTopic())) {
                                 if (next.getValue().getTagsSet().contains(msg.getTags()) //
-                                        || next.getValue().getTagsSet().contains("*")) {
+                                        || next.getValue().getTagsSet().contains("*")//
+                                        || next.getValue().getTagsSet().isEmpty()//
+                                ) {
 
                                 }
                                 else {
@@ -801,7 +805,7 @@ public List<MessageTrack> messageTrackDetail(MessageExt msg) throws RemotingExce
                         }
                     }
                     else {
-                        mt.setTrackType(TrackType.SUBSCRIBED_AND_NOT_CONSUME);
+                        mt.setTrackType(TrackType.SUBSCRIBED_AND_NOT_CONSUME_YET);
                     }
                     break;
                 default:

rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/api/TrackType.java

@@ -8,7 +8,7 @@ public enum TrackType {
     // 订阅了，但是是PULL，结果未知
     SUBSCRIBED_BUT_PULL,
     // 订阅了，但是没有消费（Offset小）
-    SUBSCRIBED_AND_NOT_CONSUME,
+    SUBSCRIBED_AND_NOT_CONSUME_YET,
     // 未知异常
     UNKNOW_EXCEPTION,
 }

rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java

@@ -1,4 +1,5 @@
 /**
+
  * Copyright (C) 2010-2013 Alibaba Group Holding Limited
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -35,6 +36,7 @@
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
 import com.alibaba.rocketmq.tools.admin.api.MessageTrack;
+import com.alibaba.rocketmq.tools.command.MQAdminStartup;
 import com.alibaba.rocketmq.tools.command.SubCommand;
 
 
@@ -218,4 +220,9 @@ private static String createBodyFile(MessageExt msg) throws IOException {
                 dos.close();
         }
     }
+
+
+    public static void main(String[] args) {
+        MQAdminStartup.main(new String[] { "queryMsgById", "-i", "0AEBAA0500002AE5000002A693AE8961" });
+    }
 }