[fix] [client] Unclear error message when creating a consumer with two same topics (#22255)

(cherry picked from commit c616b35)

Author: poorbarcode

pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java

@@ -26,9 +26,11 @@
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import com.google.common.collect.Lists;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -42,6 +44,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import lombok.Cleanup;
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.impl.ClientBuilderImpl;
 import org.apache.pulsar.client.impl.ConsumerImpl;
@@ -50,6 +53,7 @@
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.awaitility.Awaitility;
 import org.mockito.AdditionalAnswers;
 import org.mockito.Mockito;
@@ -372,6 +376,29 @@ public void testMultipleIOThreads() throws PulsarAdminException, PulsarClientExc
         assertTrue(consumer.isConnected());
     }
 
+    @Test
+    public void testSameTopics() throws Exception {
+        final String topic1 = BrokerTestUtil.newUniqueName("public/default/tp");
+        final String topic2 = "persistent://" + topic1;
+        admin.topics().createNonPartitionedTopic(topic2);
+        // Create consumer with two same topics.
+        try {
+            pulsarClient.newConsumer(Schema.INT32).topics(Arrays.asList(topic1, topic2))
+                    .subscriptionName("s1").subscribe();
+            fail("Do not allow use two same topics.");
+        } catch (Exception e) {
+            if (e instanceof PulsarClientException && e.getCause() != null) {
+                e = (Exception) e.getCause();
+            }
+            Throwable unwrapEx = FutureUtil.unwrapCompletionException(e);
+            assertTrue(unwrapEx instanceof IllegalArgumentException);
+            assertTrue(e.getMessage().contains( "Subscription topics include duplicate items"
+                    + " or invalid names"));
+        }
+        // cleanup.
+        admin.topics().delete(topic2);
+    }
+
     @Test(timeOut = 30000)
     public void testSubscriptionNotFound() throws PulsarAdminException, PulsarClientException {
         final var topic1 = newTopicName();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

@@ -165,7 +165,8 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
             return;
         }
 
-        checkArgument(topicNamesValid(conf.getTopicNames()), "Topics is invalid.");
+        checkArgument(topicNamesValid(conf.getTopicNames()), "Subscription topics include duplicate items"
+                + " or invalid names.");
 
         List<CompletableFuture<Void>> futures = conf.getTopicNames().stream()
                 .map(t -> subscribeAsync(t, createTopicIfDoesNotExist))
@@ -202,21 +203,21 @@ private static boolean topicNamesValid(Collection<String> topics) {
         checkState(topics != null && topics.size() >= 1,
             "topics should contain more than 1 topic");
 
-        Optional<String> result = topics.stream()
-                .filter(topic -> !TopicName.isValid(topic))
-                .findFirst();
+        Set<TopicName> topicNames = new HashSet<>();
 
-        if (result.isPresent()) {
-            log.warn("Received invalid topic name: {}", result.get());
-            return false;
+        for (String topic : topics) {
+            if (!TopicName.isValid(topic)) {
+                log.warn("Received invalid topic name: {}", topic);
+                return false;
+            }
+            topicNames.add(TopicName.get(topic));
         }
 
         // check topic names are unique
-        HashSet<String> set = new HashSet<>(topics);
-        if (set.size() == topics.size()) {
+        if (topicNames.size() == topics.size()) {
             return true;
         } else {
-            log.warn("Topic names not unique. unique/all : {}/{}", set.size(), topics.size());
+            log.warn("Topic names not unique. unique/all : {}/{}", topicNames.size(), topics.size());
             return false;
         }
     }