[ISSSUE 1188]Fix the problem when more than one producer or consumer in the same process can trace only one

[ahuazhu]

What is the purpose of the change

Fix the bug when there are more than one producer (or consumer) , the Tracer can not show all trace infomation.

ISSUE 1188

Brief changelog

Set different ProducerGroup for different TraceProducer, the ProducerGroup will like _INNER_TRACE_PRODUCER-user-specified-producer-group

Verifying this change

Produce and consume message for a same topic in the same process, the console show both produce and consume information.

[coveralls]

Coverage increased (+0.1%) to 50.589% when pulling 64051f2 on ahuazhu:trace_multi_topic into d66243c on apache:enchanced_msg_trace.

[zongtanghu]

Hi @duhenglucky @ShannonDing @jonnxu please help to review this pr,together.

[zongtanghu]

I think it will be better if the variable "producerOrConsumerGroup" change to be "group" name.

[zongtanghu]

The same as above mentioned.

[zongtanghu]

It will be better if makeGroupNameForTrace method name change to be genGroupNameForTrace

[ahuazhu]

All variables were renamed as your advice.

[zongtanghu]

If users start the mutiple producer or consumer instances in the same process,the instance name is the same one.You can set different instance name in the start method when the traceProducer is started. @ahuazhu

[superheizai]

LGTM

[ahuazhu]

If users start the mutiple producer or consumer instances in the same process,the instance name is the same one.You can set different instance name in the start method when the traceProducer is started. @ahuazhu

If set different instance name for trace producer, multiple client instances will be created.

[jonnxu]

if it can trace when the multi consumers or producers with the same group name?

[ChaosYjh]

if it can trace when the multi consumers or producers with the same group name?
Only the first traceProducer can trace, and the other will be start failed. If specify a instanceName for each group, it will create multi client instance.

[ahuazhu]

@jonnxu
Thank you for the reminder. As far as I know, in RocketMQ two producers for different topics can't share the same ProduceGroup, and consumers also. But a consumer and a producer can have the same group. Unfortunately，the trace will failure in such scenes. To fix this problem, client type(consume or produce) will be appended to the trace-group to distinguish the consumer and producer.

 private String genGroupNameForTrace() {
        return TraceConstants.GROUP_NAME_PREFIX + "-" + this.group + "-" + this.type ;
    }

When construct AsyncTraceDispatcher.type will be set as "PRODUCE" or "CONSUME".

This PR has been updated.

And my test code as below:

public class SameProcessSameTopicTest {
    private static String topic1 = "TopicTest1";
    private static String topic2 = "TopicTest2";

//    private static final String CONSUMER_GROUP1 = "test_c_group1";
//    private static final String CONSUMER_GROUP2 = "test_c_group2";
//
//    private static final String PRODUCER_GROUP1 = "test_p_group1";
//    private static final String PRODUCER_GROUP2 = "test_p_group2";

    private static final String CONSUMER_GROUP1 = "test__group1";
    private static final String CONSUMER_GROUP2 = "test__group2";

    private static final String PRODUCER_GROUP1 = "test__group1";
    private static final String PRODUCER_GROUP2 = "test__group2";

    private static final String NAMR_SRV = "";

    public static void startConsume(final String id, final String group, final String topic) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group, true);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe(topic, "*");
        consumer.setNamesrvAddr(NAMR_SRV);
        consumer.setInstanceName(id);
        final AtomicInteger i = new AtomicInteger(0);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new Date() + ":" + new String(msg.getBody()) + " consume success! " + msg.getMsgId() + "  " + topic);
                }
                i.getAndAdd(msgs.size());

                System.err.println(id + " Consumed " + i.get() + " messages for " + topic);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }

    public static void startProduce(String id, String group, String topic) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(group,true);
        producer.setNamesrvAddr(NAMR_SRV);
        producer.start();
        producer.setInstanceName(id);
        AtomicInteger c = new AtomicInteger(0);
        for (int i = 0; i < 10 ;) {
            try {
                Message msg = new Message(topic /* Topic */, null /* Tag */,
                        ("trace message " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                producer.send(msg);
                Thread.sleep(50);
                i++;
                System.out.println(id + " send " + c.getAndIncrement() + " messages to " + topic);
            } catch (Exception e) {
//                e.printStackTrace();
            }
        }

    }


    public static void main(String[] args) throws InterruptedException, MQClientException {

        Thread t1 = new Thread() {
            @Override
            public void run() {
                try {
                    startConsume("C1", CONSUMER_GROUP1, topic1);
                } catch (MQClientException e) {
                    e.printStackTrace();
                }
            }
        };

        Thread t2 = new Thread() {
            @Override
            public void run() {
                try {
                    startConsume("C2", CONSUMER_GROUP2, topic2);
                } catch (MQClientException e) {
                    e.printStackTrace();
                }
            }
        };


        Thread t3 = new Thread() {
            @Override
            public void run() {
                try {
                    startProduce("P1", PRODUCER_GROUP1, topic1);
                } catch (MQClientException e) {
                    e.printStackTrace();
                }
            }
        };
//
        Thread t4 = new Thread() {
            @Override
            public void run() {
                try {
                    startProduce("P2", PRODUCER_GROUP2, topic2);
                } catch (MQClientException e) {
                    e.printStackTrace();
                }
            }
        };


        t1.start();
        t2.start();
        t3.start();
        t4.start();

        t1.join();
        t2.join();
        t3.join();
        t4.join();

        System.out.printf("Consumer Started.%n");
    }
} 

[duhenglucky]

LGTM

[chenyf3]

@duhenglucky when release？Thanks！