Repository: apache/pulsar-client-go
Pull request #1156 (2024-01-12 08:10:09)

[fix] Fix DLQ producer name conflicts when multiples consumers send messages to DLQ

### Motivation

To keep consistent with the Java client.

Releted PR: https://github.com/apache/pulsar/pull/21890


*Explain here the context, and why you're making that change. What is the problem you're trying to solve.*

### Modifications

Set DLQ producerName `fmt.Sprintf("%s-%s-%s-DLQ", r.topicName, r.subscriptionName, r.consumerName)`

### 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: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / 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


crossoverJie: @RobertIndie PTAL

Patch:
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index d701ab1..75d839b 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, options.Topic, options.SubscriptionName, client.log)
+	dlq, err := newDlqRouter(client, options.DLQ, options.Topic, options.SubscriptionName, options.Name, client.log)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index 5b9314b..6be35d7 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -34,16 +34,18 @@ type dlqRouter struct {
 	closeCh          chan interface{}
 	topicName        string
 	subscriptionName string
+	consumerName     string
 	log              log.Logger
 }
 
-func newDlqRouter(client Client, policy *DLQPolicy, topicName, subscriptionName string,
+func newDlqRouter(client Client, policy *DLQPolicy, topicName, subscriptionName, consumerName string,
 	logger log.Logger) (*dlqRouter, error) {
 	r := &dlqRouter{
 		client:           client,
 		policy:           policy,
 		topicName:        topicName,
 		subscriptionName: subscriptionName,
+		consumerName:     consumerName,
 		log:              logger,
 	}
 
@@ -159,7 +161,7 @@ func (r *dlqRouter) getProducer(schema Schema) Producer {
 		opt.Topic = r.policy.DeadLetterTopic
 		opt.Schema = schema
 		if opt.Name == "" {
-			opt.Name = fmt.Sprintf("%s-%s-DLQ", r.topicName, r.subscriptionName)
+			opt.Name = fmt.Sprintf("%s-%s-%s-DLQ", r.topicName, r.subscriptionName, r.consumerName)
 		}
 
 		// the origin code sets to LZ4 compression with no options
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 0999e88..7b260b8 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, options.Topic, options.SubscriptionName, client.log)
+	dlq, err := newDlqRouter(client, nil, options.Topic, options.SubscriptionName, options.Name, 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 3e5f1d6..ebd7e4e 100644
--- a/pulsar/consumer_regex_test.go
+++ b/pulsar/consumer_regex_test.go
@@ -152,9 +152,10 @@ func runRegexConsumerDiscoverPatternAll(t *testing.T, c Client, namespace string
 	opts := ConsumerOptions{
 		SubscriptionName:    "regex-sub",
 		AutoDiscoveryPeriod: 5 * time.Minute,
+		Name:                "regex-consumer",
 	}
 
-	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", log.DefaultNopLogger())
+	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", "regex-consumer", 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 {
@@ -190,9 +191,10 @@ func runRegexConsumerDiscoverPatternFoo(t *testing.T, c Client, namespace string
 	opts := ConsumerOptions{
 		SubscriptionName:    "regex-sub",
 		AutoDiscoveryPeriod: 5 * time.Minute,
+		Name:                "regex-consumer",
 	}
 
-	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", log.DefaultNopLogger())
+	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", "regex-consumer", 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 8b983d0..df70b0d 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -1449,13 +1449,15 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 	if prodOpt != nil {
 		dlqPolicy.ProducerOptions = *prodOpt
 	}
-	sub := "my-sub"
+	sub, consumerName := "my-sub", "my-consumer"
+
 	consumer, err := client.Subscribe(ConsumerOptions{
 		Topic:               topic,
 		SubscriptionName:    sub,
 		NackRedeliveryDelay: 1 * time.Second,
 		Type:                Shared,
 		DLQ:                 &dlqPolicy,
+		Name:                consumerName,
 	})
 	assert.Nil(t, err)
 	defer consumer.Close()
@@ -1508,7 +1510,7 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 		assert.Equal(t, []byte(expectMsg), msg.Payload())
 
 		// check dql produceName
-		assert.Equal(t, msg.ProducerName(), fmt.Sprintf("%s-%s-DLQ", topic, sub))
+		assert.Equal(t, msg.ProducerName(), fmt.Sprintf("%s-%s-%s-DLQ", topic, sub, consumerName))
 
 		// check original messageId
 		assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])