produce/consume with 1 partitioned topic (#4883)

Motivation
In PR #4764, we allow to create partitioned topic with 1 partition, But in Pulsar Client, user still not able to do it.
This fix try to make sure user could create consumer/producer for 1 partitioned topic .

Modifications
- old and new added unit test passed.

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

@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.api;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.spy;
@@ -26,6 +27,7 @@ import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
@@ -3117,4 +3119,82 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
         consumer5.close();
         log.info("-- Exiting {} test --", methodName);
     }
+
+    /**
+     * This test verifies Producer and Consumer of PartitionedTopic with 1 partition works well.
+     *
+     * <pre>
+     * 1. create producer/consumer with both original name and PARTITIONED_TOPIC_SUFFIX.
+     * 2. verify producer/consumer could produce/consume messages from same underline persistent topic.
+     * </pre>
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testPartitionedTopicWithOnePartition() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+
+        final String topicName = "persistent://my-property/my-ns/one-partitioned-topic";
+        final String subscriptionName = "my-sub-";
+
+        // create partitioned topic
+        admin.topics().createPartitionedTopic(topicName, 1);
+        assertEquals(admin.topics().getPartitionedTopicMetadata(topicName).partitions, 1);
+
+        @Cleanup
+        Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+            .topic(topicName)
+            .subscriptionName(subscriptionName + 1)
+            .consumerName("aaa")
+            .subscribe();
+        log.info("Consumer1 created. topic: {}", consumer1.getTopic());
+
+        @Cleanup
+        Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+            .topic(topicName + PARTITIONED_TOPIC_SUFFIX + 0)
+            .subscriptionName(subscriptionName + 2)
+            .consumerName("bbb")
+            .subscribe();
+        log.info("Consumer2 created. topic: {}", consumer2.getTopic());
+
+        @Cleanup
+        Producer<byte[]> producer1 = pulsarClient.newProducer()
+            .topic(topicName)
+            .enableBatching(false)
+            .create();
+        log.info("Producer1 created. topic: {}", producer1.getTopic());
+
+        @Cleanup
+        Producer<byte[]> producer2 = pulsarClient.newProducer()
+            .topic(topicName + PARTITIONED_TOPIC_SUFFIX + 0)
+            .enableBatching(false)
+            .create();
+        log.info("Producer2 created. topic: {}", producer2.getTopic());
+
+        final int numMessages = 10;
+        for (int i = 0; i < numMessages; i++) {
+            producer1.newMessage()
+                .value(("one-partitioned-topic-value-producer1-" + i).getBytes(UTF_8))
+                .send();
+
+            producer2.newMessage()
+                .value(("one-partitioned-topic-value-producer2-" + i).getBytes(UTF_8))
+                .send();
+        }
+
+        for (int i = 0; i < numMessages * 2; i++) {
+            Message<byte[]> msg = consumer1.receive(200, TimeUnit.MILLISECONDS);
+            assertNotNull(msg);
+            log.info("Consumer1 Received message '{}'.", new String(msg.getValue(), UTF_8));
+
+            msg = consumer2.receive(200, TimeUnit.MILLISECONDS);
+            assertNotNull(msg);
+            log.info("Consumer2 Received message '{}'.", new String(msg.getValue(), UTF_8));
+        }
+
+        assertNull(consumer1.receive(200, TimeUnit.MILLISECONDS));
+        assertNull(consumer2.receive(200, TimeUnit.MILLISECONDS));
+
+        log.info("-- Exiting {} test --", methodName);
+    }
 }

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

@@ -745,7 +745,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
         }
 
         List<CompletableFuture<Consumer<T>>> futureList;
-        if (numPartitions > 1) {
+        if (numPartitions > 0) {
             this.topics.putIfAbsent(topicName, numPartitions);
             allTopicPartitionsNumber.addAndGet(numPartitions);
 

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

@@ -255,7 +255,7 @@ public class PulsarClientImpl implements PulsarClient {
             }
 
             ProducerBase<T> producer;
-            if (metadata.partitions > 1) {
+            if (metadata.partitions > 0) {
                 producer = new PartitionedProducerImpl<>(PulsarClientImpl.this, topic, conf, metadata.partitions,
                         producerCreatedFuture, schema, interceptors);
             } else {
@@ -342,7 +342,7 @@ public class PulsarClientImpl implements PulsarClient {
             ConsumerBase<T> consumer;
             // gets the next single threaded executor from the list of executors
             ExecutorService listenerThread = externalExecutorProvider.getExecutor();
-            if (metadata.partitions > 1) {
+            if (metadata.partitions > 0) {
                 consumer = MultiTopicsConsumerImpl.createPartitionedConsumer(PulsarClientImpl.this, conf,
                     listenerThread, consumerSubscribedFuture, metadata.partitions, schema, interceptors);
             } else {
@@ -469,7 +469,7 @@ public class PulsarClientImpl implements PulsarClient {
                 log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions);
             }
 
-            if (metadata.partitions > 1) {
+            if (metadata.partitions > 0) {
                 readerFuture.completeExceptionally(
                         new PulsarClientException("Topic reader cannot be created on a partitioned topic"));
                 return;
@@ -655,7 +655,7 @@ public class PulsarClientImpl implements PulsarClient {
     @Override
     public CompletableFuture<List<String>> getPartitionsForTopic(String topic) {
         return getPartitionedTopicMetadata(topic).thenApply(metadata -> {
-            if (metadata.partitions > 1) {
+            if (metadata.partitions > 0) {
                 TopicName topicName = TopicName.get(topic);
                 List<String> partitions = new ArrayList<>(metadata.partitions);
                 for (int i = 0; i < metadata.partitions; i++) {