Deduplicate user provided topics before using them

[wuYin]

Issue

If user provided retry topic is same with consumer topic

• Currently underlayer will create 2 same topic consumers to implement retry, but it's unnecessary, client-java only create 1 consumer to do this, see ConsumerBuilderImpl.java#L139
• When user invoke consumer.Close(), only the 2th consumer will be closed directly, consumer 1st is still active, there are potential memory leak.

Reproducation

func main() {
	client, _ := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
	defer client.Close()

	topic := "topic-01"
	consumer, _ := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            topic,
		SubscriptionName: "my-sub",
		Type:             pulsar.Shared,
		RetryEnable:      true,
		DLQ: &pulsar.DLQPolicy{
			RetryLetterTopic: topic,
			MaxDeliveries:    2,
		},
	})
	consumer.Close()
}

• issue: consumer 1 created but not close directly, it will be closed before client closing.

• log

  INFO[0000] [Connected consumer]  consumerID=1 name=pzxfg subscription=my-sub topic="persistent://public/default/topic-01"
  INFO[0000] [Created consumer]    consumerID=1 name=pzxfg subscription=my-sub topic="persistent://public/default/topic-01"
  INFO[0000] [Connected consumer]  consumerID=2 name=pzxfg subscription=my-sub topic="persistent://public/default/topic-01"
  INFO[0000] [Created consumer]    consumerID=2 name=pzxfg subscription=my-sub topic="persistent://public/default/topic-01"
  INFO[0000] Closing consumer=[2]  consumerID=2 name=pzxfg subscription=my-sub topic="persistent://public/default/topic-01"
  INFO[0000] [Closed consumer]     consumerID=2 name=pzxfg subscription=my-sub topic="persistent://public/default/topic-01"
  ---end
  

Cause

MultiTopicConsumer is using map[topic]consumer to manage consumers, but in the above case, only 2th consumer will be
corresponded with the topic, the 1st dosen't since map overwriting, it means 1st isn't managed by multiTopicConsumer , see consumer_multitopic.go#L66

Motivation

• Deduplicate user topics before using them, just like client-java's TreeSet topics implementation, see ConsumerConfigurationData.java#L56

Modifications

• add distinct utility function to deduplicate user provided topics

Verifying this change

• Make sure that the change passes the CI checks.

[wolfstudy]

LGTM +1