Copilot commented on code in PR #1417:
URL: https://github.com/apache/pulsar-client-go/pull/1417#discussion_r2319504134
##########
pulsar/consumer.go:
##########
@@ -75,6 +75,14 @@ type DLQPolicy struct {
// DeadLetterTopic specifies the name of the topic where the failing
messages will be sent.
DeadLetterTopic string
+ // DeadLetterTopicProducerName specifies a name for the producer
specifically for the DQL topic.
Review Comment:
There's a typo in the comment: 'DQL' should be 'DLQ' (Dead Letter Queue).
```suggestion
// DeadLetterTopicProducerName specifies a name for the producer
specifically for the DLQ topic.
```
##########
pulsar/consumer_test.go:
##########
@@ -1853,6 +1853,181 @@ func TestDeadLetterTopicWithInitialSubscription(t
*testing.T) {
}
+func TestWithoutDeadLetterTopicDeadLetterTopicProducerName(t *testing.T) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topic := "persistent://public/default/" + newTopicName()
+ subscriptionName := "default"
+ consumerName := "my-consumer"
+
+ dlqTopic := fmt.Sprintf("%s-%s-DLQ", topic, subscriptionName)
+ rlqTopic := fmt.Sprintf("%s-%s-RLQ", topic, subscriptionName)
+
+ producerName := "producer-name"
+ RLQProducerName := "rlq-producer-name"
+ DLQProducerName := "dlq-producer-name"
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: subscriptionName,
+ NackRedeliveryDelay: 1 * time.Millisecond,
+ Type: Shared,
+ DLQ: &DLQPolicy{
+ MaxDeliveries: 1,
+ RetryLetterTopic: rlqTopic,
+ DeadLetterTopic: dlqTopic,
+ DeadLetterTopicProducerName: DLQProducerName,
+ ProducerOptions: ProducerOptions{
+ Topic: rlqTopic,
+ Name: RLQProducerName,
+ },
+ },
+ Name: consumerName,
+ RetryEnable: true,
+ })
+ assert.Nil(t, err)
+ defer consumer.Close()
+
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topic,
+ Name: producerName,
+ })
+ assert.Nil(t, err)
+ defer producer.Close()
+
+ _, err = producer.Send(ctx, &ProducerMessage{
+ Payload: []byte("hello-0"),
+ })
+ assert.Nil(t, err)
+
+ // Validate the name of the original producer
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.NotNil(t, msg)
+ assert.Equal(t, msg.ProducerName(), producerName)
+ consumer.ReconsumeLater(msg, 0)
+
+ // Validate the name of the RLQ producer
+ msg, err = consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.NotNil(t, msg)
+ assert.Equal(t, msg.ProducerName(), RLQProducerName)
+ consumer.Nack(msg)
+
+ // Create DLQ consumer
+ dlqConsumer, err := client.Subscribe(ConsumerOptions{
+ Topic: dlqTopic,
+ SubscriptionName: subscriptionName,
+ })
+ assert.Nil(t, err)
+ defer dlqConsumer.Close()
+
+ // Validate the name of the DLQ producer
+ msg, err = dlqConsumer.Receive(ctx)
+ defer dlqConsumer.Nack(msg)
+
+ assert.Nil(t, err)
+ assert.NotNil(t, msg)
+ assert.Nil(t, err)
+ assert.Equal(t, msg.ProducerName(), DLQProducerName)
+}
+
+func TestWithDeadLetterTopicDeadLetterTopicProducerName(t *testing.T) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topic := "persistent://public/default/" + newTopicName()
+ subscriptionName := "default"
+ consumerName := "my-consumer"
+
+ dlqTopic := fmt.Sprintf("%s-%s-DLQ", topic, subscriptionName)
+ rlqTopic := fmt.Sprintf("%s-%s-RLQ", topic, subscriptionName)
+
+ producerName := "producer-name"
+ RLQProducerName := "rlq-producer-name"
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: subscriptionName,
+ NackRedeliveryDelay: 1 * time.Millisecond,
+ Type: Shared,
+ DLQ: &DLQPolicy{
+ MaxDeliveries: 1,
+ RetryLetterTopic: rlqTopic,
+ DeadLetterTopic: dlqTopic,
+ // Set no producer name for the DLQ explicitly
+ DeadLetterTopicProducerName: "",
+ ProducerOptions: ProducerOptions{
+ Topic: rlqTopic,
+ Name: RLQProducerName,
+ },
+ },
+ Name: consumerName,
+ RetryEnable: true,
+ })
+ assert.Nil(t, err)
+ defer consumer.Close()
+
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topic,
+ Name: producerName,
+ })
+ assert.Nil(t, err)
+ defer producer.Close()
+
+ _, err = producer.Send(ctx, &ProducerMessage{
+ Payload: []byte("hello-0"),
+ })
+ assert.Nil(t, err)
+
+ // Validate the name of the original producer
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.NotNil(t, msg)
+ assert.Equal(t, msg.ProducerName(), producerName)
+ consumer.ReconsumeLater(msg, 0)
+
+ // Validate the name of the RLQ producer
+ msg, err = consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.NotNil(t, msg)
+ assert.Equal(t, msg.ProducerName(), RLQProducerName)
+ consumer.Nack(msg)
+
+ // Create DLQ consumer
+ dlqConsumer, err := client.Subscribe(ConsumerOptions{
+ Topic: dlqTopic,
+ SubscriptionName: subscriptionName,
+ })
+ assert.Nil(t, err)
+ defer dlqConsumer.Close()
+
+ // Validate the name of the DLQ producer
+ msg, err = dlqConsumer.Receive(ctx)
+ defer dlqConsumer.Nack(msg)
+
+ assert.Nil(t, err)
+ assert.NotNil(t, msg)
+ assert.Nil(t, err)
Review Comment:
Line 2026 is redundant - the error is already checked on line 2024. This
duplicate assertion should be removed.
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
