[fix][broker] Fix creation of replicated subscriptions for partitioned topics (#24997)

Author: lhotari

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -2220,7 +2220,7 @@ protected void internalCreateSubscription(AsyncResponse asyncResponse, String su
                                     try {
                                         pulsar().getAdminClient().topics()
                                                 .createSubscriptionAsync(topicNamePartition.toString(),
-                                                        subscriptionName, targetMessageId, false, properties)
+                                                        subscriptionName, targetMessageId, replicated, properties)
                                                 .handle((r, ex) -> {
                                                     if (ex != null) {
                                                         // fail the operation on unknown exception or

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

@@ -2543,6 +2543,27 @@ public void testInvalidBundleErrorResponse() throws Exception {
         }
     }
 
+    @Test(dataProvider = "trueFalse")
+    public void testCreateReplicatedSubscriptionForPartitionedTopic(boolean replicated) throws Exception {
+        final String topic = newUniqueName("persistent://" + defaultNamespace + "/topic");
+        admin.topics().createPartitionedTopic(topic, 10);
+        admin.topics().createSubscription(topic, "sub", MessageId.earliest, replicated);
+        for (int i = 0; i < 10; i++) {
+            String individualPartition = TopicName.get(topic).getPartition(i).toString();
+            TopicStats stats = admin.topics().getStats(individualPartition);
+            assertEquals(stats.getSubscriptions().get("sub").isReplicated(), replicated);
+        }
+    }
+
+    @Test(dataProvider = "trueFalse")
+    public void testCreateReplicatedSubscriptionForNonPartitionedTopic(boolean replicated) throws Exception {
+        final String topic = newUniqueName("persistent://" + defaultNamespace + "/topic");
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().createSubscription(topic, "sub", MessageId.earliest, replicated);
+        TopicStats stats = admin.topics().getStats(topic);
+        assertEquals(stats.getSubscriptions().get("sub").isReplicated(), replicated);
+    }
+
     @Test
     public void testMaxSubscriptionsPerTopic() throws Exception {
         restartClusterAfterTest();