Enabling RocketMQ Message Tracing and Querying Traces

Do as follows to enable message tracing on clients:

• Enabling message tracing on a producer client (tracing messages other than transactional messages)

  Set enableMsgTrace of the constructor to true. For example:

  DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName", true);

• Enabling message tracing on a producer client (tracing transactional messages)

  Set enableMsgTrace of the constructor to true. For example:

  TransactionMQProducer producer = new TransactionMQProducer(null, "ProducerGroupName", null, true, null);

• Enabling message tracing on a consumer

  Set enableMsgTrace of the constructor to true. For example:

  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName", true);

Do as follows to enable message tracing on clients:

1. Run the following command to check whether Go has been installed:

   go version

   If the following information is displayed, Go has been installed. If Go is not installed, download and install it.

   [root@ecs-test sarama]# go version
   go version go1.16.5 linux/amd64

2. Go to the bin directory where the Go script is in.
3. Run the touch go.mod command to create a go.mod file and add the following code to it to add the dependency:

   module rocketmq-example-go
   go 1.13
   require (
   	github.com/apache/rocketmq-client-go/v2 v2.1.0
   )

4. Enable message tracing on the producer. Replace the information in bold with the actual values.

   package main
   
   import (
   	"context"
   	"fmt"
   	"os"
   	"time"
   
   	"github.com/apache/rocketmq-client-go/v2"
   	"github.com/apache/rocketmq-client-go/v2/primitive"
   	"github.com/apache/rocketmq-client-go/v2/producer"
   )
   
   func main() {
   	namesrvs := []string{"192.168.0.1:8100"}
   	traceCfg := &primitive.TraceConfig{
   		Access:   primitive.Local,
   		Resolver: primitive.NewPassthroughResolver(namesrvs),
   	}
   
   	p, _ := rocketmq.NewProducer(
   		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
   		producer.WithRetry(2),
   		producer.WithTrace(traceCfg))   // To enable message tracing, add this line.
   	err := p.Start()
   	if err != nil {
   		fmt.Printf("start producer error: %s", err.Error())
   		os.Exit(1)
   	}
   	res, err := p.SendSync(context.Background(), primitive.NewMessage("topic1",
   		[]byte("Hello RocketMQ Go Client!")))
   
   	if err != nil {
   		fmt.Printf("send message error: %s\n", err)
   	} else {
   		fmt.Printf("send message success: result=%s\n", res.String())
   	}
   
   	time.Sleep(10 * time.Second)
   
   	err = p.Shutdown()
   	if err != nil {
   		fmt.Printf("shutdown producer error: %s", err.Error())
   	}
   }

   192.168.0.1:8100 is the connection address and port of the RocketMQ instance. Obtain yours from Overview > Connection on the RocketMQ console.

5. Enable message tracing on the consumer. Replace the information in bold with the actual values.

   package main
   
   import (
   	"context"
   	"fmt"
   	"os"
   	"time"
   
   	"github.com/apache/rocketmq-client-go/v2"
   	"github.com/apache/rocketmq-client-go/v2/consumer"
   	"github.com/apache/rocketmq-client-go/v2/primitive"
   )
   
   func main() {
   	namesrvs := []string{"192.168.0.1:8100"}
   	traceCfg := &primitive.TraceConfig{
   		Access:   primitive.Local,
   		Resolver: primitive.NewPassthroughResolver(namesrvs),
   	}
   
   	c, _ := rocketmq.NewPushConsumer(
   		consumer.WithGroupName("testGroup"),
   		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
   		consumer.WithTrace(traceCfg),   // To enable message tracing, add this line.
   	)
   	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
   		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
   		fmt.Printf("subscribe callback: %v \n", msgs)
   		return consumer.ConsumeSuccess, nil
   	})
   	if err != nil {
   		fmt.Println(err.Error())
   	}
   	// Note: start after subscribe
   	err = c.Start()
   	if err != nil {
   		fmt.Println(err.Error())
   		os.Exit(-1)
   
   	}
   	time.Sleep(time.Hour)
   	err = c.Shutdown()
   	if err != nil {
   		fmt.Printf("shutdown Consumer error: %s", err.Error())
   	}
   }

   To obtain the values in bold in the preceding code, see Table 1.

   Table 1 Obtaining the values in bold

   Value	Description	How to Obtain
   192.168.0.1:8100	Connection address and port of the RocketMQ instance	Overview > Connection on the RocketMQ console.
   testGroup	Name of the consumer group created in the RocketMQ instance	Instance > Consumer Groups page on the RocketMQ console.
   TopicTest	Name of the topic created in the RocketMQ instance	Instance > Topics page on the RocketMQ console.

Do as follows to enable message tracing on clients:

• For producers

  Add the following line in the application.properties configuration file:

  rocketmq.producer.enable-msg-trace=true

• For consumers

  Set parameter enableMsgTrace to true. For example:

  @Service
  @RocketMQMessageListener(
      topic = "test-topic-1", 
      consumerGroup = "my-consumer_test-topic-1",
      enableMsgTrace = true
  )
  public class MyConsumer implements RocketMQListener<String> {
      ...
  }

  To obtain the values in bold in the preceding code, see Table 1.

For more information, see Message Tracing. To configure parameter accessChannel, use the default value LOCAL.