Repository: apache/pulsar-client-go
Pull request #1137 (2023-11-22 01:30:22)

[Improve] Set dlq producerName


### Motivation

To keep consistent with the Java client.

Releted PR: https://github.com/apache/pulsar/pull/21589

*Explain here the context, and why you're making that change. What is the problem you're trying to solve.*

### Modifications

Set DLQ producerName `%s-%s-DLQ`

### Verifying this change

- [x] Make sure that the change passes the CI checks.

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (no)
  - The public API: (no)
  - The schema: (no)
  - The default values of configurations: (yes)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index d782bea..d701ab1 100644
--- a/pulsar/consumer_impl.go
+++ b/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
 	}
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index 6d30003..5b9314b 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -19,6 +19,7 @@ package pulsar
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
@@ -26,19 +27,24 @@ import (
 )
 
 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
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index ffc92de..0999e88 100644
--- a/pulsar/reader_impl.go
+++ b/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
 	}


Test patch:
diff --git a/pulsar/consumer_regex_test.go b/pulsar/consumer_regex_test.go
index 9cb600f..de94613 100644
--- a/pulsar/consumer_regex_test.go
+++ b/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 {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 2d785b4..33382cf 100644
--- a/pulsar/consumer_test.go
+++ b/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])