[fix][broker] Partitioned shadow topic not work properly

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2198,8 +2198,13 @@ protected CompletableFuture<Void> addShadowReplicationCluster(String shadowTopic
                 .thenAccept(replicationClient -> {
                     Replicator replicator = shadowReplicators.computeIfAbsent(shadowTopic, r -> {
                         try {
-                            return new ShadowReplicator(shadowTopic, PersistentTopic.this, cursor, brokerService,
-                                    (PulsarClientImpl) replicationClient);
+                            TopicName sourceTopicName = TopicName.get(getName());
+                            String shadowPartitionTopic = shadowTopic;
+                            if (sourceTopicName.isPartitioned()) {
+                                shadowPartitionTopic += "-partition-" + sourceTopicName.getPartitionIndex();
+                            }
+                            return new ShadowReplicator(shadowPartitionTopic, PersistentTopic.this, cursor,
+                                    brokerService, (PulsarClientImpl) replicationClient);
                         } catch (PulsarServerException e) {
                             log.error("[{}] ShadowReplicator startup failed {}", topic, shadowTopic, e);
                         }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java

@@ -21,6 +21,8 @@
 
 import com.google.common.collect.Lists;
 import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import lombok.AllArgsConstructor;
 import lombok.Cleanup;
@@ -113,6 +115,35 @@ public void testPartitionedShadowTopicSetup() throws Exception {
         Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopicPartition);
     }
 
+    @Test
+    public void testPartitionedShadowTopicProduceAndConsume() throws Exception {
+        String sourceTopic = newShadowSourceTopicName();
+        String shadowTopic = sourceTopic + "-shadow";
+        admin.topics().createPartitionedTopic(sourceTopic, 3);
+        admin.topics().createShadowTopic(shadowTopic, sourceTopic);
+
+        admin.topics().setShadowTopics(sourceTopic, Lists.newArrayList(shadowTopic));
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(sourceTopic).create();
+        @Cleanup
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(shadowTopic).subscriptionName("test")
+                .subscribe();
+
+        for (int i = 0; i < 10; i++) {
+            producer.send("msg-" + i);
+        }
+
+        Set<String> set = new HashSet<>();
+        for (int i = 0; i < 10; i++) {
+            Message<String> msg = consumer.receive();
+            set.add(msg.getValue());
+        }
+        for (int i = 0; i < 10; i++) {
+            Assert.assertTrue(set.contains("msg-" + i));
+        }
+    }
+
     @Test
     public void testShadowTopicNotWritable() throws Exception {
         String sourceTopic = newShadowSourceTopicName();