Merge pull request #649 from zongtanghu/msg_track

[ISSUE#525]Add a switch variable(traceTopicEnable) for making msg trace worked which is decided by users and rename enableAcl flag variable.

acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionLoader.java

@@ -119,7 +119,6 @@ public class PlainPermissionLoader {
                                     && (StandardWatchEventKinds.ENTRY_MODIFY.equals(event.kind())
                                     || StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind()))) {
                                     log.info("{} make a difference  change is : {}", watchFileName, event.toString());
-                                    //TODO do a retain
                                     //Clearing the info, may result in a non-available time
                                     PlainPermissionLoader.this.clearPermissionInfo();
                                     initialize();

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -492,7 +492,7 @@ public class BrokerController {
     }
 
     private void initialAcl() {
-        if (!this.brokerConfig.isEnableAcl()) {
+        if (!this.brokerConfig.isAclEnable()) {
             log.info("The broker dose not enable acl");
             return;
         }

broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java

@@ -125,12 +125,14 @@ public class TopicConfigManager extends ConfigManager {
             this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
         }
         {
-            String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
-            TopicConfig topicConfig = new TopicConfig(topic);
-            this.systemTopicList.add(topic);
-            topicConfig.setReadQueueNums(1);
-            topicConfig.setWriteQueueNums(1);
-            this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
+            if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {
+                String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
+                TopicConfig topicConfig = new TopicConfig(topic);
+                this.systemTopicList.add(topic);
+                topicConfig.setReadQueueNums(1);
+                topicConfig.setWriteQueueNums(1);
+                this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
+            }
         }
     }
 

common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java

@@ -53,6 +53,8 @@ public class BrokerConfig {
     private String messageStorePlugIn = "";
     @ImportantField
     private String msgTraceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
+    @ImportantField
+    private boolean traceTopicEnable = false;
     /**
      * thread numbers for send message thread pool, since spin lock will be used by default since 4.0.x, the default
      * value is 1.
@@ -176,8 +178,7 @@ public class BrokerConfig {
      * Acl feature switch
      */
     @ImportantField
-    private boolean enableAcl = false;
-
+    private boolean aclEnable = false;
 
     public static String localHostName() {
         try {
@@ -717,14 +718,6 @@ public class BrokerConfig {
         this.transactionCheckInterval = transactionCheckInterval;
     }
 
-    public boolean isEnableAcl() {
-        return enableAcl;
-    }
-
-    public void setEnableAcl(boolean isAclPlug) {
-        this.enableAcl = isAclPlug;
-    }
-
     public int getEndTransactionThreadPoolNums() {
         return endTransactionThreadPoolNums;
     }
@@ -756,5 +749,20 @@ public class BrokerConfig {
     public void setMsgTraceTopicName(String msgTraceTopicName) {
         this.msgTraceTopicName = msgTraceTopicName;
     }
+    
+    public boolean isTraceTopicEnable() {
+        return traceTopicEnable;
+    }
+
+    public void setTraceTopicEnable(boolean traceTopicEnable) {
+        this.traceTopicEnable = traceTopicEnable;
+    }
 
+    public boolean isAclEnable() {
+        return aclEnable;
+    }
+
+    public void setAclEnable(boolean aclEnable) {
+        this.aclEnable = aclEnable;
+    }
 }