[Improve] Set dlq producerName

pulsar/consumer_impl.go

@@ -167,7 +167,7 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 		}
 	}
 
-	dlq, err := newDlqRouter(client, options.DLQ, client.log)
+	dlq, err := newDlqRouter(client, options.DLQ, options.Topic, options.SubscriptionName, client.log)
 	if err != nil {
 		return nil, err
 	}

pulsar/consumer_regex_test.go

@@ -154,7 +154,7 @@ func runRegexConsumerDiscoverPatternAll(t *testing.T, c Client, namespace string
 		AutoDiscoveryPeriod: 5 * time.Minute,
 	}
 
-	dlq, _ := newDlqRouter(c.(*client), nil, log.DefaultNopLogger())
+	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", log.DefaultNopLogger())
 	rlq, _ := newRetryRouter(c.(*client), nil, false, log.DefaultNopLogger())
 	consumer, err := newRegexConsumer(c.(*client), opts, tn, pattern, make(chan ConsumerMessage, 1), dlq, rlq)
 	if err != nil {
@@ -192,7 +192,7 @@ func runRegexConsumerDiscoverPatternFoo(t *testing.T, c Client, namespace string
 		AutoDiscoveryPeriod: 5 * time.Minute,
 	}
 
-	dlq, _ := newDlqRouter(c.(*client), nil, log.DefaultNopLogger())
+	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", log.DefaultNopLogger())
 	rlq, _ := newRetryRouter(c.(*client), nil, false, log.DefaultNopLogger())
 	consumer, err := newRegexConsumer(c.(*client), opts, tn, pattern, make(chan ConsumerMessage, 1), dlq, rlq)
 	if err != nil {

pulsar/consumer_test.go

@@ -1449,9 +1449,10 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 	if prodOpt != nil {
 		dlqPolicy.ProducerOptions = *prodOpt
 	}
+	sub := "my-sub"
 	consumer, err := client.Subscribe(ConsumerOptions{
 		Topic:               topic,
-		SubscriptionName:    "my-sub",
+		SubscriptionName:    sub,
 		NackRedeliveryDelay: 1 * time.Second,
 		Type:                Shared,
 		DLQ:                 &dlqPolicy,
@@ -1506,6 +1507,9 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 		expectMsg := fmt.Sprintf("hello-%d", expectedMsgIdx)
 		assert.Equal(t, []byte(expectMsg), msg.Payload())
 
+		// check dql produceName
+		assert.Equal(t, msg.ProducerName(), fmt.Sprintf("%s-%s-DLQ", topic, sub))
+
 		// check original messageId
 		assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
 

pulsar/dlq_router.go

@@ -19,26 +19,32 @@ package pulsar
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"github.com/apache/pulsar-client-go/pulsar/log"
 )
 
 type dlqRouter struct {
-	client    Client
-	producer  Producer
-	policy    *DLQPolicy
-	messageCh chan ConsumerMessage
-	closeCh   chan interface{}
-	log       log.Logger
+	client           Client
+	producer         Producer
+	policy           *DLQPolicy
+	messageCh        chan ConsumerMessage
+	closeCh          chan interface{}
+	topicName        string
+	subscriptionName string
+	log              log.Logger
 }
 
-func newDlqRouter(client Client, policy *DLQPolicy, logger log.Logger) (*dlqRouter, error) {
+func newDlqRouter(client Client, policy *DLQPolicy, topicName, subscriptionName string,
+	logger log.Logger) (*dlqRouter, error) {
 	r := &dlqRouter{
-		client: client,
-		policy: policy,
-		log:    logger,
+		client:           client,
+		policy:           policy,
+		topicName:        topicName,
+		subscriptionName: subscriptionName,
+		log:              logger,
 	}
 
 	if policy != nil {
@@ -152,6 +158,9 @@ func (r *dlqRouter) getProducer(schema Schema) Producer {
 		opt := r.policy.ProducerOptions
 		opt.Topic = r.policy.DeadLetterTopic
 		opt.Schema = schema
+		if opt.Name == "" {
+			opt.Name = fmt.Sprintf("%s-%s-DLQ", r.topicName, r.subscriptionName)
+		}
 
 		// the origin code sets to LZ4 compression with no options
 		// so the new design allows compression type to be overwritten but still set lz4 by default

pulsar/reader_impl.go

@@ -127,7 +127,7 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 	}
 
 	// Provide dummy dlq router with not dlq policy
-	dlq, err := newDlqRouter(client, nil, client.log)
+	dlq, err := newDlqRouter(client, nil, options.Topic, options.SubscriptionName, client.log)
 	if err != nil {
 		return nil, err
 	}