[ISSUE #2412] Support to add TAG when producing message

example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java

@@ -41,6 +41,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.srvutil.ServerUtil;
 
 public class Producer {
+
     public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {
 
         Options options = ServerUtil.buildCommandlineOptions(new Options());
@@ -54,11 +55,12 @@ public class Producer {
         final int messageSize = commandLine.hasOption('s') ? Integer.parseInt(commandLine.getOptionValue('s')) : 128;
         final boolean keyEnable = commandLine.hasOption('k') && Boolean.parseBoolean(commandLine.getOptionValue('k'));
         final int propertySize = commandLine.hasOption('p') ? Integer.parseInt(commandLine.getOptionValue('p')) : 0;
+        final int tagCount = commandLine.hasOption('l') ? Integer.parseInt(commandLine.getOptionValue('l')) : 0;
         final boolean msgTraceEnable = commandLine.hasOption('m') && Boolean.parseBoolean(commandLine.getOptionValue('m'));
         final boolean aclEnable = commandLine.hasOption('a') && Boolean.parseBoolean(commandLine.getOptionValue('a'));
 
-        System.out.printf("topic %s threadCount %d messageSize %d keyEnable %s traceEnable %s aclEnable %s%n",
-            topic, threadCount, messageSize, keyEnable, msgTraceEnable, aclEnable);
+        System.out.printf("topic: %s threadCount: %d messageSize: %d keyEnable: %s propertySize: %d tagCount: %d traceEnable: %s aclEnable: %s%n",
+            topic, threadCount, messageSize, keyEnable, propertySize, tagCount, msgTraceEnable, aclEnable);
 
         final InternalLogger log = ClientLogger.getLog();
 
@@ -134,6 +136,10 @@ public class Producer {
                             if (keyEnable) {
                                 msg.setKeys(String.valueOf(beginTimestamp / 1000));
                             }
+                            if (tagCount > 0) {
+                                long sendSucCount = statsBenchmark.getReceiveResponseSuccessCount().get();
+                                msg.setTags(String.format("tag%d", sendSucCount % tagCount));
+                            }
                             if (propertySize > 0) {
                                 if (msg.getProperties() != null) {
                                     msg.getProperties().clear();
@@ -215,6 +221,10 @@ public class Producer {
         opt.setRequired(false);
         options.addOption(opt);
 
+        opt = new Option("l", "tagCount", true, "Tag count, Default: 0");
+        opt.setRequired(false);
+        options.addOption(opt);
+
         opt = new Option("m", "msgTraceEnable", true, "Message Trace Enable, Default: false");
         opt.setRequired(false);
         options.addOption(opt);