Fix missing topic in dlq producer name when using RetryEnable option

[geniusjoe]

Fixes #1314
Master Issue: apache/pulsar#21589

Motivation

When creating a consumer with option.RetryEnable option, consumer initialization needs to subscribe both option.Topic and options.DLQ.RetryLetterTopic, it will set option.Topic to empty string and move this topic to option.Topics for subscribing. Codes below:

pulsar-client-go/pulsar/consumer_impl.go

Line 176 in 7a9a33c

if options.Topic != "" && len(options.Topics) == 0 {

	if options.RetryEnable {
		... omit code

		if options.Topic != "" && len(options.Topics) == 0 {
			options.Topics = []string{options.Topic, options.DLQ.RetryLetterTopic} # move options.Topic to options.Topics 
			options.Topic = ""
		} else if options.Topic == "" && len(options.Topics) > 0 {
			options.Topics = append(options.Topics, options.DLQ.RetryLetterTopic)  # move options.Topic to options.Topics 
		}
	}

But newDlqRouter() function use option.Topic when generating default producer name, when we set option.RetryEnable to true, default producer name will become to -<subscriptionName>-<consumerName>-<randomString>-DLQ (lose topicName part).

pulsar-client-go/pulsar/consumer_impl.go

Line 184 in 7a9a33c

dlq, err := newDlqRouter(client, options.DLQ, options.Topic, options.SubscriptionName, options.Name,

Modifications

Modify pulsar/consumer_impl.go#newDlqRouter() initialization, when enable option.RetryEnable option, we try to find source topic in option.Topics. And if we set option.RetryEnable to false or there are multiple source topics, keep consistent with previous strategy(i.e. empty string).

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as
pulsar/consumer_test.go#TestRLQ
pulsar/consumer_test.go#TestRLQMultiTopics

Does this pull request potentially affect one of the following parts:

• 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)