I have finished developing the new feature for the message track alib…

…aba#525 initially.
CST11021 · Nov 13, 2018 · e04db85 · e04db85
1 parent 37223ca
commit e04db85
Show file tree

Hide file tree

Showing 55 changed files with 1,827 additions and 51 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -228,6 +228,14 @@ private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext
             return response;
         }
 
+        if (this.brokerController.getBrokerConfig().isAutoTraceBrokerEnable()) {
+            String errorMsg = "the topic[" + requestHeader.getTopic() + "] is user self defined topic and this node is trace broker!";
+            log.warn(errorMsg);
+            response.setCode(ResponseCode.SYSTEM_ERROR);
+            response.setRemark(errorMsg);
+            return response;
+        }
+
         try {
             response.setCode(ResponseCode.SUCCESS);
             response.setOpaque(request.getOpaque());

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
@@ -124,6 +124,17 @@ public TopicConfigManager(BrokerController brokerController) {
             topicConfig.setWriteQueueNums(1);
             this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
         }
+        {
+            // MixAll.RMQ_SYS_TRACK_TRACE_TOPIC
+            if (this.brokerController.getBrokerConfig().isAutoTraceBrokerEnable()) {
+                String topic = MixAll.RMQ_SYS_TRACK_TRACE_TOPIC;
+                TopicConfig topicConfig = new TopicConfig(topic);
+                this.systemTopicList.add(topic);
+                topicConfig.setReadQueueNums(1);
+                topicConfig.setWriteQueueNums(1);
+                this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
+            }
+        }
     }
 
     public boolean isSystemTopic(final String topic) {

diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
@@ -18,6 +18,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.QueryResult;
@@ -29,13 +30,20 @@
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.trace.core.common.TrackTraceConstants;
+import org.apache.rocketmq.client.trace.core.common.TrackTraceDispatcherType;
+import org.apache.rocketmq.client.trace.core.dispatch.AsyncDispatcher;
+import org.apache.rocketmq.client.trace.core.dispatch.impl.AsyncArrayDispatcher;
+import org.apache.rocketmq.client.trace.core.hook.ConsumeMessageTraceHookImpl;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 
@@ -56,6 +64,8 @@
  */
 public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
 
+    private final InternalLogger log = ClientLogger.getLog();
+
     /**
      * Internal implementation. Most of the functions herein are delegated to it.
      */
@@ -246,11 +256,16 @@ public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsume
      */
     private long consumeTimeout = 15;
 
+    /**
+     * Interface of asynchronous transfer data
+     */
+    private AsyncDispatcher traceDispatcher = null;
+
     /**
      * Default constructor.
      */
     public DefaultMQPushConsumer() {
-        this(MixAll.DEFAULT_CONSUMER_GROUP, null, new AllocateMessageQueueAveragely());
+        this(MixAll.DEFAULT_CONSUMER_GROUP, null, new AllocateMessageQueueAveragely(), false);
     }
 
     /**
@@ -261,10 +276,29 @@ public DefaultMQPushConsumer() {
      * @param allocateMessageQueueStrategy message queue allocating algorithm.
      */
     public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
-        AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
+        AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean msgTraceSwitch) {
         this.consumerGroup = consumerGroup;
         this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
         defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
+        //if client open the message track trace feature
+        if (msgTraceSwitch) {
+            try {
+                Properties tempProperties = new Properties();
+                tempProperties.put(TrackTraceConstants.MAX_MSG_SIZE, "128000");
+                tempProperties.put(TrackTraceConstants.ASYNC_BUFFER_SIZE, "2048");
+                tempProperties.put(TrackTraceConstants.MAX_BATCH_NUM, "100");
+                tempProperties.put(TrackTraceConstants.INSTANCE_NAME, "PID_CLIENT_INNER_TRACE_PRODUCER");
+                tempProperties.put(TrackTraceConstants.TRACE_DISPATCHER_TYPE, TrackTraceDispatcherType.CONSUMER.name());
+                AsyncArrayDispatcher dispatcher = new AsyncArrayDispatcher(tempProperties);
+                dispatcher.setHostConsumer(this.getDefaultMQPushConsumerImpl());
+                traceDispatcher = dispatcher;
+
+                this.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(
+                    new ConsumeMessageTraceHookImpl(traceDispatcher));
+            } catch (Throwable e) {
+                log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
+            }
+        }
     }
 
     /**
@@ -273,16 +307,16 @@ public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
      * @param rpcHook RPC hook to execute before each remoting command.
      */
     public DefaultMQPushConsumer(RPCHook rpcHook) {
-        this(MixAll.DEFAULT_CONSUMER_GROUP, rpcHook, new AllocateMessageQueueAveragely());
+        this(MixAll.DEFAULT_CONSUMER_GROUP, rpcHook, new AllocateMessageQueueAveragely(),false);
     }
 
     /**
      * Constructor specifying consumer group.
      *
      * @param consumerGroup Consumer group.
      */
-    public DefaultMQPushConsumer(final String consumerGroup) {
-        this(consumerGroup, null, new AllocateMessageQueueAveragely());
+    public DefaultMQPushConsumer(final String consumerGroup, boolean msgTraceSwitch) {
+        this(consumerGroup, null, new AllocateMessageQueueAveragely(),msgTraceSwitch);
     }
 
     @Override
@@ -518,6 +552,15 @@ public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClie
     @Override
     public void start() throws MQClientException {
         this.defaultMQPushConsumerImpl.start();
+        if (null != traceDispatcher) {
+            try {
+                Properties tempProperties = new Properties();
+                tempProperties.put(TrackTraceConstants.NAMESRV_ADDR, this.getNamesrvAddr());
+                traceDispatcher.start(tempProperties);
+            } catch (MQClientException e) {
+                log.warn("trace dispatcher start failed ", e);
+            }
+        }
     }
 
     /**
@@ -526,6 +569,9 @@ public void start() throws MQClientException {
     @Override
     public void shutdown() {
         this.defaultMQPushConsumerImpl.shutdown();
+        if (null != traceDispatcher) {
+            traceDispatcher.shutdown();
+        }
     }
 
     @Override
@@ -694,4 +740,12 @@ public long getConsumeTimeout() {
     public void setConsumeTimeout(final long consumeTimeout) {
         this.consumeTimeout = consumeTimeout;
     }
+
+    public AsyncDispatcher getTraceDispatcher() {
+        return traceDispatcher;
+    }
+
+    public void setTraceDispatcher(AsyncDispatcher traceDispatcher) {
+        this.traceDispatcher = traceDispatcher;
+    }
 }
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
@@ -144,7 +144,7 @@ public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String cli
 
         this.rebalanceService = new RebalanceService(this);
 
-        this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
+        this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP,false);
         this.defaultMQProducer.resetClientConfig(clientConfig);
 
         this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);

diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
@@ -18,13 +18,20 @@
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Properties;
 import java.util.concurrent.ExecutorService;
 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.QueryResult;
 import org.apache.rocketmq.client.Validators;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.trace.core.common.TrackTraceConstants;
+import org.apache.rocketmq.client.trace.core.common.TrackTraceDispatcherType;
+import org.apache.rocketmq.client.trace.core.dispatch.AsyncDispatcher;
+import org.apache.rocketmq.client.trace.core.dispatch.impl.AsyncArrayDispatcher;
+import org.apache.rocketmq.client.trace.core.hook.SendMessageTrackHookImpl;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageBatch;
@@ -33,6 +40,7 @@
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageId;
 import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
@@ -56,6 +64,8 @@
  */
 public class DefaultMQProducer extends ClientConfig implements MQProducer {
 
+    private final InternalLogger log = ClientLogger.getLog();
+
     /**
      * Wrapping internal implementations for virtually all methods presented in this class.
      */
@@ -119,40 +129,65 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
      */
     private int maxMessageSize = 1024 * 1024 * 4; // 4M
 
+    /**
+     * Interface of asynchronous transfer data
+     */
+    private AsyncDispatcher traceDispatcher = null;
+
     /**
      * Default constructor.
      */
     public DefaultMQProducer() {
-        this(MixAll.DEFAULT_PRODUCER_GROUP, null);
+        this(MixAll.DEFAULT_PRODUCER_GROUP, null,false);
     }
 
     /**
      * Constructor specifying both producer group and RPC hook.
      *
      * @param producerGroup Producer group, see the name-sake field.
      * @param rpcHook RPC hook to execute per each remoting command execution.
+     * @param msgTraceSwitch switch flag instance for message track trace
      */
-    public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) {
+    public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean msgTraceSwitch) {
         this.producerGroup = producerGroup;
         defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
+        //if client open the message track trace feature
+        if (msgTraceSwitch) {
+            try {
+                Properties tempProperties = new Properties();
+                tempProperties.put(TrackTraceConstants.MAX_MSG_SIZE, "128000");
+                tempProperties.put(TrackTraceConstants.ASYNC_BUFFER_SIZE, "2048");
+                tempProperties.put(TrackTraceConstants.MAX_BATCH_NUM, "100");
+                tempProperties.put(TrackTraceConstants.INSTANCE_NAME, "PID_CLIENT_INNER_TRACE_PRODUCER");
+                tempProperties.put(TrackTraceConstants.TRACE_DISPATCHER_TYPE, TrackTraceDispatcherType.PRODUCER.name());
+                AsyncArrayDispatcher dispatcher = new AsyncArrayDispatcher(tempProperties);
+                dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
+                traceDispatcher = dispatcher;
+
+                this.getDefaultMQProducerImpl().registerSendMessageHook(
+                    new SendMessageTrackHookImpl(traceDispatcher));
+            } catch (Throwable e) {
+                log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
+            }
+        }
     }
 
     /**
      * Constructor specifying producer group.
      *
      * @param producerGroup Producer group, see the name-sake field.
      */
-    public DefaultMQProducer(final String producerGroup) {
-        this(producerGroup, null);
+    public DefaultMQProducer(final String producerGroup, boolean msgTraceSwitch) {
+        this(producerGroup, null,msgTraceSwitch);
     }
 
     /**
      * Constructor specifying the RPC hook.
-     *
+     * 
      * @param rpcHook RPC hook to execute per each remoting command execution.
      */
     public DefaultMQProducer(RPCHook rpcHook) {
-        this(MixAll.DEFAULT_PRODUCER_GROUP, rpcHook);
+        this(MixAll.DEFAULT_PRODUCER_GROUP, rpcHook,false);
     }
 
     /**
@@ -170,6 +205,15 @@ public DefaultMQProducer(RPCHook rpcHook) {
     @Override
     public void start() throws MQClientException {
         this.defaultMQProducerImpl.start();
+        if (null != traceDispatcher) {
+            try {
+                Properties tempProperties = new Properties();
+                tempProperties.put(TrackTraceConstants.NAMESRV_ADDR, this.getNamesrvAddr());
+                traceDispatcher.start(tempProperties);
+            } catch (MQClientException e) {
+                log.warn("trace dispatcher start failed ", e);
+            }
+        }
     }
 
     /**
@@ -178,6 +222,9 @@ public void start() throws MQClientException {
     @Override
     public void shutdown() {
         this.defaultMQProducerImpl.shutdown();
+        if (null != traceDispatcher) {
+            traceDispatcher.shutdown();
+        }
     }
 
     /**
@@ -777,4 +824,12 @@ public int getRetryTimesWhenSendAsyncFailed() {
     public void setRetryTimesWhenSendAsyncFailed(final int retryTimesWhenSendAsyncFailed) {
         this.retryTimesWhenSendAsyncFailed = retryTimesWhenSendAsyncFailed;
     }
+
+    public AsyncDispatcher getTraceDispatcher() {
+        return traceDispatcher;
+    }
+
+    public void setTraceDispatcher(AsyncDispatcher traceDispatcher) {
+        this.traceDispatcher = traceDispatcher;
+    }
 }
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/TransactionMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/TransactionMQProducer.java
@@ -35,11 +35,11 @@ public TransactionMQProducer() {
     }
 
     public TransactionMQProducer(final String producerGroup) {
-        super(producerGroup);
+        super(producerGroup,false);
     }
 
     public TransactionMQProducer(final String producerGroup, RPCHook rpcHook) {
-        super(producerGroup, rpcHook);
+        super(producerGroup, rpcHook,false);
     }
 
     @Override