Repository: apache/pulsar-client-go (go)

Issue:
InitialSubscriptionName for DLQPolicy
**Is your feature request related to a problem? Please describe.**
[From the docs](https://pulsar.apache.org/docs/3.3.x/concepts-messaging/#dead-letter-topic):

>By default, there is no subscription during DLQ topic creation. Without a just-in-time subscription to the DLQ topic, you may lose messages. To automatically create an initial subscription for the DLQ, you can specify the `initialSubscriptionName` parameter.

These docs reference the java client. I don't see a corresponding option in the go client ([I would expect to find a corresponding field on the `DLQPolicy` struct](https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#DLQPolicy)).

**Describe the solution you'd like**
The ability to create an initial subscription for a DLQ when creating a consumer with a retry-letter/deadletter policy. Likely a new `string` or `*string` field on `DLQPolicy` which does the same thing as `initialSubscriptionName()` from the java client does, but only if the value is not empty

**Describe alternatives you've considered**
I believe the alternative in order to avoid losing messages is to initialize the subscription to the dead letter topic before creating the consumer which produces to the dead letter topic.

If this option already exists, please point me in the right direction and close this issue.

Thanks!

Pull request description:
[Improve] Add InitialSubscriptionName for DLQPolicy


Fixes #1239



### Modifications

Add `InitialSubscriptionName` for DLQPolicy.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*

### 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: (yes)
  - The schema: (no)
  - The default values of configurations: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (GoDocs)
  - 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.go b/pulsar/consumer.go
index 31f89a5..bf2eafb 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -78,6 +78,12 @@ type DLQPolicy struct {
 
 	// RetryLetterTopic specifies the name of the topic where the retry messages will be sent.
 	RetryLetterTopic string
+
+	// InitialSubscriptionName Name of the initial subscription name of the dead letter topic.
+	// If this field is not set, the initial subscription for the dead letter topic will not be created.
+	// If this field is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer
+	// will fail to be created.
+	InitialSubscriptionName string
 }
 
 // AckGroupingOptions controls how to group ACK requests
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index 6be35d7..647c022 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -163,6 +163,7 @@ func (r *dlqRouter) getProducer(schema Schema) Producer {
 		if opt.Name == "" {
 			opt.Name = fmt.Sprintf("%s-%s-%s-DLQ", r.topicName, r.subscriptionName, r.consumerName)
 		}
+		opt.initialSubscriptionName = r.policy.InitialSubscriptionName
 
 		// 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/producer.go b/pulsar/producer.go
index f8013a1..0ae51bd 100644
--- a/pulsar/producer.go
+++ b/pulsar/producer.go
@@ -207,6 +207,12 @@ type ProducerOptions struct {
 	// - ProducerAccessModeShared
 	// - ProducerAccessModeExclusive
 	ProducerAccessMode
+
+	// initialSubscriptionName Name of the initial subscription name of the dead letter topic.
+	// If this field is not set, the initial subscription for the dead letter topic will not be created.
+	// If this field is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer
+	// will fail to be created.
+	initialSubscriptionName string
 }
 
 // Producer is used to publish messages on a topic
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index f5fd493..1677c57 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -273,6 +273,7 @@ func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
 		Epoch:                    proto.Uint64(atomic.LoadUint64(&p.epoch)),
 		UserProvidedProducerName: proto.Bool(p.userProvidedProducerName),
 		ProducerAccessMode:       toProtoProducerAccessMode(p.options.ProducerAccessMode).Enum(),
+		InitialSubscriptionName:  proto.String(p.options.initialSubscriptionName),
 	}
 
 	if p.topicEpoch != nil {


Test patch:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 2ecdf8e..04439cb 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -1539,6 +1539,91 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 	assert.Error(t, err)
 	assert.Nil(t, msg)
 }
+func TestDeadLetterTopicWithInitialSubscription(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := "persistent://public/default/" + newTopicName()
+	dlqSub, sub, consumerName := "init-sub", "my-sub", "my-consumer"
+	dlqTopic := fmt.Sprintf("%s-%s-DLQ", topic, sub)
+	ctx := context.Background()
+
+	// create consumer
+	maxRedeliveryCount, sendMessages := 1, 100
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:               topic,
+		SubscriptionName:    sub,
+		NackRedeliveryDelay: 1 * time.Second,
+		Type:                Shared,
+		DLQ: &DLQPolicy{
+			MaxDeliveries:           uint32(maxRedeliveryCount),
+			DeadLetterTopic:         dlqTopic,
+			InitialSubscriptionName: dlqSub,
+		},
+		Name:              consumerName,
+		ReceiverQueueSize: sendMessages,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topic,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// send messages
+	for i := 0; i < sendMessages; i++ {
+		if _, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		}); err != nil {
+			log.Fatal(err)
+		}
+	}
+
+	// nack all messages
+	for i := 0; i < sendMessages*(maxRedeliveryCount+1); i++ {
+		ctx, canc := context.WithTimeout(context.Background(), 3*time.Second)
+		defer canc()
+		msg, _ := consumer.Receive(ctx)
+		if msg == nil {
+			break
+		}
+		consumer.Nack(msg)
+	}
+
+	// create dlq consumer
+	dlqConsumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            dlqTopic,
+		SubscriptionName: dlqSub,
+	})
+	assert.Nil(t, err)
+	defer dlqConsumer.Close()
+
+	for i := 0; i < sendMessages; i++ {
+		ctx, canc := context.WithTimeout(context.Background(), 100*time.Millisecond)
+		defer canc()
+		msg, err := dlqConsumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.NotNil(t, msg)
+		err = dlqConsumer.Ack(msg)
+		assert.Nil(t, err)
+	}
+
+	// No more messages on the DLQ
+	ctx, canc := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer canc()
+	msg, err := dlqConsumer.Receive(ctx)
+	assert.Error(t, err)
+	assert.Nil(t, msg)
+
+}
 
 func TestDLQMultiTopics(t *testing.T) {
 	client, err := NewClient(ClientOptions{