[fix][broker] Fix REST API to produce messages to single-partitioned topics (#24450)

Author: Joforde

pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java

@@ -345,7 +345,7 @@ private boolean findOwnerBrokerForTopic(boolean authoritative, AsyncResponse asy
         List<String> redirectAddresses = Collections.synchronizedList(new ArrayList<>());
         CompletableFuture<Boolean> future = new CompletableFuture<>();
         List<CompletableFuture<Void>> lookupFutures = new ArrayList<>();
-        if (!topicName.isPartitioned() && metadata.partitions > 1) {
+        if (!topicName.isPartitioned() && metadata.partitions > 0) {
             // Partitioned topic with multiple partitions, need to do look up for each partition.
             for (int index = 0; index < metadata.partitions; index++) {
                 lookupFutures.add(lookUpBrokerForTopic(topicName.getPartition(index),

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java

@@ -96,6 +96,7 @@
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 public class TopicsTest extends MockedPulsarServiceBaseTest {
@@ -168,10 +169,19 @@ private static List<ProducerMessage> createMessages(String message) throws JsonP
                 }).readValue(message);
     }
 
-    @Test
-    public void testProduceToPartitionedTopic() throws Exception {
+    @DataProvider(name = "partitionNumbers")
+    public Object[][] partitionNumbers() {
+        return new Object[][] {
+            //produce to single-partitioned topic
+            {1},
+            {5},
+        };
+    }
+
+    @Test(dataProvider = "partitionNumbers")
+    public void testProduceToPartitionedTopic(int numPartitions) throws Exception {
         admin.topics().createPartitionedTopic("persistent://" + testTenant + "/" + testNamespace
-                + "/" + testTopicName + "-p", 5);
+                + "/" + testTopicName + "-p", numPartitions);
         AsyncResponse asyncResponse = mock(AsyncResponse.class);
         Schema<String> schema = StringSchema.utf8();
         ProducerMessages producerMessages = new ProducerMessages();
@@ -210,7 +220,7 @@ public void testProduceToPartitionedTopic() throws Exception {
         }
         for (int index = 0; index < messagePerPartition.length; index++) {
             // We publish to each partition in round robin mode so each partition should get at most 2 message.
-            Assert.assertTrue(messagePerPartition[index] <= 2);
+            Assert.assertTrue(messagePerPartition[index] <= 10 / numPartitions);
         }
     }