diff --git a/conf/broker.conf b/conf/broker.conf
index 59626ed884dd8..27d3190a8c22e 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1279,6 +1279,10 @@ exposePreciseBacklogInPrometheus=false
 
 splitTopicAndPartitionLabelInPrometheus=false
 
+# If true and the client supports partial producer, aggregate publisher stats of PartitionedTopicStats by producerName.
+# Otherwise, aggregate it by list index.
+aggregatePublisherStatsByProducerName=false
+
 ### --- Schema storage --- ###
 # The schema storage implementation used by this broker
 schemaRegistryStorageClassName=org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorageFactory
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 9a727177df7ba..a60f024033716 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -907,6 +907,10 @@ exposePreciseBacklogInPrometheus=false
 
 splitTopicAndPartitionLabelInPrometheus=false
 
+# If true, aggregate publisher stats of PartitionedTopicStats by producerName.
+# Otherwise, aggregate it by list index.
+aggregatePublisherStatsByProducerName=false
+
 ### --- Deprecated config variables --- ###
 
 # Deprecated. Use configurationStoreServers
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index fb6a134661336..b5644351e1c61 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2322,6 +2322,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "Stats update initial delay in seconds"
     )
     private int statsUpdateInitialDelayInSecs = 60;
+    @FieldContext(
+        category = CATEGORY_METRICS,
+        doc = "If true, aggregate publisher stats of PartitionedTopicStats by producerName"
+    )
+    private boolean aggregatePublisherStatsByProducerName = false;
 
     /**** --- Ledger Offloading. --- ****/
     /****
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
index 0f94b6d1c6464..99d21db569be2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
@@ -37,6 +37,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.service.BrokerServiceException.TopicClosedException;
 import org.apache.pulsar.broker.service.BrokerServiceException.TopicTerminatedException;
 import org.apache.pulsar.broker.service.Topic.PublishContext;
@@ -98,7 +99,10 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
             boolean isEncrypted, Map<String, String> metadata, SchemaVersion schemaVersion, long epoch,
             boolean userProvidedProducerName,
             ProducerAccessMode accessMode,
-            Optional<Long> topicEpoch) {
+            Optional<Long> topicEpoch,
+            boolean supportsPartialProducer) {
+        final ServiceConfiguration serviceConf =  cnx.getBrokerService().pulsar().getConfiguration();
+
         this.topic = topic;
         this.cnx = cnx;
         this.producerId = producerId;
@@ -124,11 +128,20 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
         stats.setClientVersion(cnx.getClientVersion());
         stats.setProducerName(producerName);
         stats.producerId = producerId;
+        if (serviceConf.isAggregatePublisherStatsByProducerName()) {
+            // If true and the client supports partial producer,
+            // aggregate publisher stats of PartitionedTopicStats by producerName.
+            // Otherwise, aggregate it by list index.
+            stats.setSupportsPartialProducer(supportsPartialProducer);
+        } else {
+            // aggregate publisher stats of PartitionedTopicStats by list index.
+            stats.setSupportsPartialProducer(false);
+        }
         stats.metadata = this.metadata;
         stats.accessMode = Commands.convertProducerAccessMode(accessMode);
 
 
-        String replicatorPrefix = cnx.getBrokerService().pulsar().getConfiguration().getReplicatorPrefix() + ".";
+        String replicatorPrefix = serviceConf.getReplicatorPrefix() + ".";
         this.isRemote = producerName.startsWith(replicatorPrefix);
         this.remoteCluster = parseRemoteClusterName(producerName, isRemote, replicatorPrefix);
 
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index a605f312c6e22..0d47eeb80d714 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1141,6 +1141,7 @@ protected void handleProducer(final CommandProducer cmdProducer) {
         final Optional<Long> topicEpoch = cmdProducer.hasTopicEpoch()
                 ? Optional.of(cmdProducer.getTopicEpoch()) : Optional.empty();
         final boolean isTxnEnabled = cmdProducer.isTxnEnabled();
+        final boolean supportsPartialProducer = supportsPartialProducer();
 
         TopicName topicName = validateTopicName(cmdProducer.getTopic(), requestId, cmdProducer);
         if (topicName == null) {
@@ -1239,7 +1240,7 @@ protected void handleProducer(final CommandProducer cmdProducer) {
                         topic.checkIfTransactionBufferRecoverCompletely(isTxnEnabled).thenAccept(future -> {
                             buildProducerAndAddTopic(topic, producerId, producerName, requestId, isEncrypted,
                                     metadata, schemaVersion, epoch, userProvidedProducerName, topicName,
-                                    producerAccessMode, topicEpoch, producerFuture);
+                                    producerAccessMode, topicEpoch, supportsPartialProducer, producerFuture);
                         }).exceptionally(exception -> {
                             Throwable cause = exception.getCause();
                             log.error("producerId {}, requestId {} : TransactionBuffer recover failed",
@@ -1305,11 +1306,12 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ
                              boolean isEncrypted, Map<String, String> metadata, SchemaVersion schemaVersion, long epoch,
                              boolean userProvidedProducerName, TopicName topicName,
                              ProducerAccessMode producerAccessMode,
-                             Optional<Long> topicEpoch, CompletableFuture<Producer> producerFuture){
+                             Optional<Long> topicEpoch, boolean supportsPartialProducer,
+                             CompletableFuture<Producer> producerFuture){
         CompletableFuture<Void> producerQueuedFuture = new CompletableFuture<>();
         Producer producer = new Producer(topic, ServerCnx.this, producerId, producerName,
                 getPrincipal(), isEncrypted, metadata, schemaVersion, epoch,
-                userProvidedProducerName, producerAccessMode, topicEpoch);
+                userProvidedProducerName, producerAccessMode, topicEpoch, supportsPartialProducer);
 
         topic.addProducer(producer, producerQueuedFuture).thenAccept(newTopicEpoch -> {
             if (isActive()) {
@@ -2715,6 +2717,10 @@ boolean supportBrokerMetadata() {
         return features != null && features.isSupportsBrokerEntryMetadata();
     }
 
+    boolean supportsPartialProducer() {
+        return features != null && features.isSupportsPartialProducer();
+    }
+
     @Override
     public String getClientVersion() {
         return clientVersion;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index e238df344c44e..566ae5f9a3677 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -824,7 +824,7 @@ public CompletableFuture<NonPersistentTopicStatsImpl> asyncGetStats(boolean getP
             if (producer.isRemote()) {
                 remotePublishersStats.put(producer.getRemoteCluster(), publisherStats);
             } else {
-                stats.getPublishers().add(publisherStats);
+                stats.addPublisher(publisherStats);
             }
         });
 
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 0b917a03dd1b6..2ba03eb3a654f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1902,7 +1902,7 @@ public CompletableFuture<TopicStatsImpl> asyncGetStats(boolean getPreciseBacklog
             if (producer.isRemote()) {
                 remotePublishersStats.put(producer.getRemoteCluster(), publisherStats);
             } else {
-                stats.publishers.add(publisherStats);
+                stats.addPublisher(publisherStats);
             }
         });
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
index 513398fb3e81e..1f31ae34d052b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -66,14 +66,17 @@
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerAccessMode;
 import org.apache.pulsar.client.api.ProxyProtocol;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.TopicMetadata;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.TopicDomain;
@@ -2222,7 +2225,6 @@ public void testGetTopicsWithDifferentMode() throws Exception {
         producer2.close();
     }
 
-
     @Test(dataProvider = "isV1")
     public void testNonPartitionedTopic(boolean isV1) throws Exception {
         String tenant = "prop-xyz";
@@ -2270,4 +2272,83 @@ public void testFailedUpdatePartitionedTopic() throws Exception {
         // validate subscription is created for new partition.
         assertNotNull(admin.topics().getStats(partitionedTopicName + "-partition-" + 6).getSubscriptions().get(subName1));
     }
+
+    @Test(dataProvider = "topicType")
+    public void testPartitionedStatsAggregationByProducerName(String topicType) throws Exception {
+        conf.setAggregatePublisherStatsByProducerName(true);
+        final String topic = topicType + "://prop-xyz/ns1/test-partitioned-stats-aggregation-by-producer-name";
+        admin.topics().createPartitionedTopic(topic, 10);
+
+        @Cleanup
+        Producer<byte[]> producer1 = pulsarClient.newProducer()
+                .topic(topic)
+                .enableLazyStartPartitionedProducers(true)
+                .enableBatching(false)
+                .messageRoutingMode(MessageRoutingMode.CustomPartition)
+                .messageRouter(new MessageRouter() {
+                    @Override
+                    public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+                        return msg.hasKey() ? Integer.parseInt(msg.getKey()) : 0;
+                    }
+                })
+                .accessMode(ProducerAccessMode.Shared)
+                .create();
+
+        @Cleanup
+        Producer<byte[]> producer2 = pulsarClient.newProducer()
+                .topic(topic)
+                .enableLazyStartPartitionedProducers(true)
+                .enableBatching(false)
+                .messageRoutingMode(MessageRoutingMode.CustomPartition)
+                .messageRouter(new MessageRouter() {
+                    @Override
+                    public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+                        return msg.hasKey() ? Integer.parseInt(msg.getKey()) : 5;
+                    }
+                })
+                .accessMode(ProducerAccessMode.Shared)
+                .create();
+
+        for (int i = 0; i < 10; i++) {
+            producer1.newMessage()
+                    .key(String.valueOf(i % 5))
+                    .value(("message".getBytes(StandardCharsets.UTF_8)))
+                    .send();
+            producer2.newMessage()
+                    .key(String.valueOf(i % 5 + 5))
+                    .value(("message".getBytes(StandardCharsets.UTF_8)))
+                    .send();
+        }
+
+        PartitionedTopicStats topicStats = admin.topics().getPartitionedStats(topic, true);
+        assertEquals(topicStats.getPartitions().size(), 10);
+        assertEquals(topicStats.getPartitions().values().stream().mapToInt(e -> e.getPublishers().size()).sum(), 10);
+        assertEquals(topicStats.getPartitions().values().stream().map(e -> e.getPublishers().get(0).getProducerName()).distinct().count(), 2);
+        assertEquals(topicStats.getPublishers().size(), 2);
+        topicStats.getPublishers().forEach(p -> assertTrue(p.isSupportsPartialProducer()));
+    }
+
+    @Test(dataProvider = "topicType")
+    public void testPartitionedStatsAggregationByProducerNamePerPartition(String topicType) throws Exception {
+        conf.setAggregatePublisherStatsByProducerName(true);
+        final String topic = topicType + "://prop-xyz/ns1/test-partitioned-stats-aggregation-by-producer-name-per-pt";
+        admin.topics().createPartitionedTopic(topic, 2);
+
+        @Cleanup
+        Producer<byte[]> producer1 = pulsarClient.newProducer()
+                .topic(topic + TopicName.PARTITIONED_TOPIC_SUFFIX + 0)
+                .create();
+
+        @Cleanup
+        Producer<byte[]> producer2 = pulsarClient.newProducer()
+                .topic(topic + TopicName.PARTITIONED_TOPIC_SUFFIX + 1)
+                .create();
+
+        PartitionedTopicStats topicStats = admin.topics().getPartitionedStats(topic, true);
+        assertEquals(topicStats.getPartitions().size(), 2);
+        assertEquals(topicStats.getPartitions().values().stream().mapToInt(e -> e.getPublishers().size()).sum(), 2);
+        assertEquals(topicStats.getPartitions().values().stream().map(e -> e.getPublishers().get(0).getProducerName()).distinct().count(), 2);
+        assertEquals(topicStats.getPublishers().size(), 2);
+        topicStats.getPublishers().forEach(p -> assertTrue(p.isSupportsPartialProducer()));
+    }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index e1b316cd821fc..bd45e275c4185 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -415,7 +415,7 @@ public void testAddRemoveProducer() throws Exception {
         // 1. simple add producer
         Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
                 role, false, null, SchemaVersion.Latest, 0, false,
-                ProducerAccessMode.Shared, Optional.empty());
+                ProducerAccessMode.Shared, Optional.empty(), true);
         topic.addProducer(producer, new CompletableFuture<>());
         assertEquals(topic.getProducers().size(), 1);
 
@@ -432,7 +432,7 @@ public void testAddRemoveProducer() throws Exception {
         PersistentTopic failTopic = new PersistentTopic(failTopicName, ledgerMock, brokerService);
         Producer failProducer = new Producer(failTopic, serverCnx, 2 /* producer id */, "prod-name",
                 role, false, null, SchemaVersion.Latest, 0, false,
-                ProducerAccessMode.Shared, Optional.empty());
+                ProducerAccessMode.Shared, Optional.empty(), true);
         try {
             topic.addProducer(failProducer, new CompletableFuture<>());
             fail("should have failed");
@@ -443,7 +443,7 @@ public void testAddRemoveProducer() throws Exception {
         // 4. Try to remove with unequal producer
         Producer producerCopy = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
                 role, false, null, SchemaVersion.Latest, 0, false,
-                ProducerAccessMode.Shared, Optional.empty());
+                ProducerAccessMode.Shared, Optional.empty(), true);
         topic.removeProducer(producerCopy);
         // Expect producer to be in map
         assertEquals(topic.getProducers().size(), 1);
@@ -462,9 +462,9 @@ public void testProducerOverwrite() throws Exception {
         PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
         String role = "appid1";
         Producer producer1 = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
-                role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty(), true);
         Producer producer2 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name",
-                role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty(), true);
         try {
             topic.addProducer(producer1, new CompletableFuture<>()).join();
             topic.addProducer(producer2, new CompletableFuture<>()).join();
@@ -477,7 +477,7 @@ public void testProducerOverwrite() throws Exception {
         Assert.assertEquals(topic.getProducers().size(), 1);
 
         Producer producer3 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name",
-                role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty(), true);
 
         try {
             topic.addProducer(producer3, new CompletableFuture<>()).join();
@@ -493,7 +493,7 @@ public void testProducerOverwrite() throws Exception {
         Assert.assertEquals(topic.getProducers().size(), 0);
 
         Producer producer4 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name",
-                role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty(), true);
 
         topic.addProducer(producer3, new CompletableFuture<>());
         topic.addProducer(producer4, new CompletableFuture<>());
@@ -506,13 +506,13 @@ public void testProducerOverwrite() throws Exception {
         Assert.assertEquals(topic.getProducers().size(), 0);
 
         Producer producer5 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1",
-                role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty(), true);
 
         topic.addProducer(producer5, new CompletableFuture<>());
         Assert.assertEquals(topic.getProducers().size(), 1);
 
         Producer producer6 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1",
-                role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty(), true);
 
         topic.addProducer(producer6, new CompletableFuture<>());
         Assert.assertEquals(topic.getProducers().size(), 1);
@@ -520,7 +520,7 @@ public void testProducerOverwrite() throws Exception {
         topic.getProducers().values().forEach(producer -> Assert.assertEquals(producer.getEpoch(), 2));
 
         Producer producer7 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1",
-                role, false, null, SchemaVersion.Latest, 3, true, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 3, true, ProducerAccessMode.Shared, Optional.empty(), true);
 
         topic.addProducer(producer7, new CompletableFuture<>());
         Assert.assertEquals(topic.getProducers().size(), 1);
@@ -533,20 +533,20 @@ private void testMaxProducers() throws Exception {
         String role = "appid1";
         // 1. add producer1
         Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name1", role,
-                false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
+                false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
         topic.addProducer(producer, new CompletableFuture<>());
         assertEquals(topic.getProducers().size(), 1);
 
         // 2. add producer2
         Producer producer2 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name2", role,
-                false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
+                false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
         topic.addProducer(producer2, new CompletableFuture<>());
         assertEquals(topic.getProducers().size(), 2);
 
         // 3. add producer3 but reached maxProducersPerTopic
         try {
             Producer producer3 = new Producer(topic, serverCnx, 3 /* producer id */, "prod-name3", role,
-                    false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
+                    false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
             topic.addProducer(producer3, new CompletableFuture<>()).join();
             fail("should have failed");
         } catch (Exception e) {
@@ -596,7 +596,7 @@ private Producer getMockedProducerWithSpecificAddress(Topic topic, long producer
         doReturn(new PulsarCommandSenderImpl(null, cnx)).when(cnx).getCommandSender();
 
         return new Producer(topic, cnx, producerId, producerNameBase + producerId, role, false, null,
-                SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
+                SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
     }
 
     @Test
@@ -1255,7 +1255,7 @@ public void testDeleteTopic() throws Exception {
         // 3. delete topic with producer
         topic = (PersistentTopic) brokerService.getOrCreateTopic(successTopicName).get();
         Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
-                role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
+                role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
         topic.addProducer(producer, new CompletableFuture<>()).join();
 
         assertTrue(topic.delete().isCompletedExceptionally());
@@ -1429,7 +1429,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
             String role = "appid1";
             Thread.sleep(10); /* delay to ensure that the delete gets executed first */
             Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
-                    role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
+                    role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
             topic.addProducer(producer, new CompletableFuture<>()).join();
             fail("Should have failed");
         } catch (Exception e) {
@@ -2190,7 +2190,7 @@ public void testDisconnectProducer() throws Exception {
         String role = "appid1";
         Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
                 role, false, null, SchemaVersion.Latest, 0, false,
-                ProducerAccessMode.Shared, Optional.empty());
+                ProducerAccessMode.Shared, Optional.empty(), true);
         assertFalse(producer.isDisconnecting());
         // Disconnect the producer multiple times.
         producer.disconnect();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
index ae097de5eab03..3504e263d515f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
@@ -144,7 +144,7 @@ public void testLargeMessage(boolean ackReceiptEnabled) throws Exception {
 
         pulsar.getBrokerService().updateRates();
 
-        PublisherStats producerStats = topic.getStats(false, false, false).publishers.get(0);
+        PublisherStats producerStats = topic.getStats(false, false, false).getPublishers().get(0);
 
         assertTrue(producerStats.getChunkedMessageRate() > 0);
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
index 8136cf07c345a..2177d1b2db520 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
@@ -22,6 +22,7 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import java.io.IOException;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -133,9 +134,10 @@ public CompletableFuture<ClientCnx> getConnection(String topic) {
     @Override
     protected <T> ProducerImpl<T> newProducerImpl(String topic, int partitionIndex, ProducerConfigurationData conf,
                                                   Schema<T> schema, ProducerInterceptors interceptors,
-                                                  CompletableFuture<Producer<T>> producerCreatedFuture) {
+                                                  CompletableFuture<Producer<T>> producerCreatedFuture,
+                                                  Optional<String> overrideProducerName) {
         return new ProducerImpl<T>(this, topic, conf, producerCreatedFuture, partitionIndex, schema,
-                interceptors) {
+                interceptors, overrideProducerName) {
             @Override
             protected OpSendMsgQueue createPendingMessagesQueue() {
                 return new OpSendMsgQueue() {
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java
index f3cf6147c0745..fb50e33abf8ed 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java
@@ -43,6 +43,9 @@ public interface PublisherStats {
     /** Id of this publisher. */
     long getProducerId();
 
+    /** Whether partial producer is supported at client. */
+    boolean isSupportsPartialProducer();
+
     /** Producer name. */
     String getProducerName();
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
index 10646af77a737..844c7fa89a826 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
@@ -34,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.pulsar.client.api.Message;
@@ -64,6 +65,7 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
     private final ProducerStatsRecorderImpl stats;
     private TopicMetadata topicMetadata;
     private final int firstPartitionIndex;
+    private String overrideProducerName;
 
     // timeout related to auto check and subscribe partition increasement
     private volatile Timeout partitionsAutoUpdateTimeout = null;
@@ -148,42 +150,67 @@ private void start(List<Integer> indexList) {
         AtomicReference<Throwable> createFail = new AtomicReference<Throwable>();
         AtomicInteger completed = new AtomicInteger();
 
-        for (int partitionIndex : indexList) {
-            createProducer(partitionIndex).producerCreatedFuture().handle((prod, createException) -> {
-                if (createException != null) {
-                    setState(State.Failed);
-                    createFail.compareAndSet(null, createException);
+        final BiConsumer<Boolean, Throwable> afterCreatingProducer = (failFast, createException) -> {
+            final Runnable closeRunnable = () -> {
+                log.error("[{}] Could not create partitioned producer.", topic, createFail.get().getCause());
+                closeAsync().handle((ok, closeException) -> {
+                    producerCreatedFuture().completeExceptionally(createFail.get());
+                    client.cleanupProducer(this);
+                    return null;
+                });
+            };
+
+            if (createException != null) {
+                setState(State.Failed);
+                createFail.compareAndSet(null, createException);
+                if (failFast) {
+                    closeRunnable.run();
                 }
-                // we mark success if all the partitions are created
-                // successfully, else we throw an exception
-                // due to any
-                // failure in one of the partitions and close the successfully
-                // created partitions
-                if (completed.incrementAndGet() == indexList.size()) {
-                    if (createFail.get() == null) {
-                        setState(State.Ready);
-                        log.info("[{}] Created partitioned producer", topic);
-                        producerCreatedFuture().complete(PartitionedProducerImpl.this);
-                    } else {
-                        log.error("[{}] Could not create partitioned producer.", topic, createFail.get().getCause());
-                        closeAsync().handle((ok, closeException) -> {
-                            producerCreatedFuture().completeExceptionally(createFail.get());
-                            client.cleanupProducer(this);
-                            return null;
-                        });
-                    }
+            }
+            // we mark success if all the partitions are created
+            // successfully, else we throw an exception
+            // due to any
+            // failure in one of the partitions and close the successfully
+            // created partitions
+            if (completed.incrementAndGet() == indexList.size()) {
+                if (createFail.get() == null) {
+                    setState(State.Ready);
+                    log.info("[{}] Created partitioned producer", topic);
+                    producerCreatedFuture().complete(PartitionedProducerImpl.this);
+                } else {
+                    closeRunnable.run();
                 }
+            }
+        };
 
-                return null;
-            });
-        }
+        final ProducerImpl<T> firstProducer = createProducer(indexList.get(0));
+        firstProducer.producerCreatedFuture().handle((prod, createException) -> {
+            afterCreatingProducer.accept(true, createException);
+            if (createException != null) {
+                throw new RuntimeException(createException);
+            }
+            overrideProducerName = firstProducer.getProducerName();
+            return Optional.of(overrideProducerName);
+        }).thenApply(name -> {
+            for (int i = 1; i < indexList.size(); i++) {
+                createProducer(indexList.get(i), name).producerCreatedFuture().handle((prod, createException) -> {
+                    afterCreatingProducer.accept(false, createException);
+                    return null;
+                });
+            }
+            return null;
+        });
     }
 
     private ProducerImpl<T> createProducer(final int partitionIndex) {
+        return createProducer(partitionIndex, Optional.empty());
+    }
+
+    private ProducerImpl<T> createProducer(final int partitionIndex, final Optional<String> overrideProducerName) {
         return producers.computeIfAbsent(partitionIndex, (idx) -> {
             String partitionName = TopicName.get(topic).getPartition(idx).toString();
             return client.newProducerImpl(partitionName, idx,
-                    conf, schema, interceptors, new CompletableFuture<>());
+                    conf, schema, interceptors, new CompletableFuture<>(), overrideProducerName);
         });
     }
 
@@ -203,7 +230,7 @@ CompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message, Transa
                 "Illegal partition index chosen by the message routing policy: " + partition);
 
         if (conf.isLazyStartPartitionedProducers() && !producers.containsKey(partition)) {
-            final ProducerImpl<T> newProducer = createProducer(partition);
+            final ProducerImpl<T> newProducer = createProducer(partition, Optional.ofNullable(overrideProducerName));
             final State createState = newProducer.producerCreatedFuture().handle((prod, createException) -> {
                 if (createException != null) {
                     log.error("[{}] Could not create internal producer. partitionIndex: {}", topic, partition,
@@ -379,7 +406,8 @@ public CompletableFuture<Void> onTopicsExtended(Collection<String> topicsExtende
                                     int partitionIndex = TopicName.getPartitionIndex(partitionName);
                                     return producers.computeIfAbsent(partitionIndex, (idx) -> new ProducerImpl<>(
                                             client, partitionName, conf, new CompletableFuture<>(),
-                                            idx, schema, interceptors)).producerCreatedFuture();
+                                            idx, schema, interceptors,
+                                            Optional.ofNullable(overrideProducerName))).producerCreatedFuture();
                                 }).collect(Collectors.toList());
 
                         FutureUtil.waitForAll(futureList)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index aec4d6796b006..7f28649edfb9b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -161,7 +161,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
 
     public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf,
                         CompletableFuture<Producer<T>> producerCreatedFuture, int partitionIndex, Schema<T> schema,
-                        ProducerInterceptors interceptors) {
+                        ProducerInterceptors interceptors, Optional<String> overrideProducerName) {
         super(client, topic, conf, producerCreatedFuture, schema, interceptors);
         this.producerId = client.newProducerId();
         this.producerName = conf.getProducerName();
@@ -173,6 +173,7 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
         } else {
             this.semaphore = Optional.empty();
         }
+        overrideProducerName.ifPresent(key -> this.producerName = key);
 
         this.compressor = CompressionCodecProvider.getCompressionCodec(conf.getCompressionType());
 
@@ -1559,7 +1560,6 @@ public void connectionOpened(final ClientCnx cnx) {
                         }
                         topicEpoch = response.getTopicEpoch();
 
-
                         if (this.producerName == null) {
                             this.producerName = producerName;
                         }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index cf0a7509cbc96..068e3dddd5e60 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -346,7 +346,8 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
                 producer = newPartitionedProducerImpl(topic, conf, schema, interceptors, producerCreatedFuture,
                         metadata);
             } else {
-                producer = newProducerImpl(topic, -1, conf, schema, interceptors, producerCreatedFuture);
+                producer = newProducerImpl(topic, -1, conf, schema, interceptors, producerCreatedFuture,
+                        Optional.empty());
             }
 
             producers.add(producer);
@@ -403,9 +404,10 @@ protected <T> ProducerImpl<T> newProducerImpl(String topic, int partitionIndex,
                                                   ProducerConfigurationData conf,
                                                   Schema<T> schema,
                                                   ProducerInterceptors interceptors,
-                                                  CompletableFuture<Producer<T>> producerCreatedFuture) {
+                                                  CompletableFuture<Producer<T>> producerCreatedFuture,
+                                                  Optional<String> overrideProducerName) {
         return new ProducerImpl<>(PulsarClientImpl.this, topic, conf, producerCreatedFuture, partitionIndex, schema,
-                interceptors);
+                interceptors, overrideProducerName);
     }
 
     public CompletableFuture<Consumer<byte[]>> subscribeAsync(ConsumerConfigurationData<byte[]> conf) {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
index 11379b36c1e62..b45f99d07cccb 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
@@ -36,6 +36,7 @@
 import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadFactory;
@@ -80,12 +81,12 @@ public void setup() {
         when(client.getConfiguration()).thenReturn(clientConfigurationData);
         when(client.timer()).thenReturn(timer);
         when(client.newProducer()).thenReturn(producerBuilderImpl);
-        when(client.newProducerImpl(anyString(), anyInt(), any(), any(), any(), any()))
+        when(client.newProducerImpl(anyString(), anyInt(), any(), any(), any(), any(), any()))
                 .thenAnswer(invocationOnMock -> {
             return new ProducerImpl<>(client, invocationOnMock.getArgument(0),
                     invocationOnMock.getArgument(2), invocationOnMock.getArgument(5),
                     invocationOnMock.getArgument(1), invocationOnMock.getArgument(3),
-                    invocationOnMock.getArgument(4));
+                    invocationOnMock.getArgument(4), invocationOnMock.getArgument(6));
         });
     }
 
@@ -259,7 +260,7 @@ public void testGetNumOfPartitions() throws Exception {
         String nonPartitionedTopicName = "test-get-num-of-partitions-for-non-partitioned-topic";
         ProducerConfigurationData producerConfDataNonPartitioned = new ProducerConfigurationData();
         ProducerImpl producerImpl = new ProducerImpl(clientImpl, nonPartitionedTopicName, producerConfDataNonPartitioned,
-                null, 0, null, null);
+                null, 0, null, null, Optional.empty());
         assertEquals(producerImpl.getNumOfPartitions(), 0);
     }
 
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentTopicStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentTopicStatsImpl.java
index b95ccd82e6ceb..35c665f881d7c 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentTopicStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/NonPersistentTopicStatsImpl.java
@@ -22,16 +22,21 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import lombok.Getter;
 import org.apache.pulsar.common.policies.data.NonPersistentPublisherStats;
 import org.apache.pulsar.common.policies.data.NonPersistentReplicatorStats;
 import org.apache.pulsar.common.policies.data.NonPersistentSubscriptionStats;
 import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
+import org.apache.pulsar.common.policies.data.PublisherStats;
 
 /**
  * Statistics for a non-persistent topic.
@@ -57,7 +62,11 @@ public class NonPersistentTopicStatsImpl extends TopicStatsImpl implements NonPe
 
     @JsonProperty("publishers")
     public List<NonPersistentPublisherStats> getNonPersistentPublishers() {
-        return (List<NonPersistentPublisherStats>) nonPersistentPublishers;
+        return Stream.concat(nonPersistentPublishers.stream()
+                                .sorted(Comparator.comparing(NonPersistentPublisherStats::getProducerName)),
+                nonPersistentPublishersMap.values().stream()
+                        .sorted(Comparator.comparing(NonPersistentPublisherStats::getProducerName)))
+                .collect(Collectors.toList());
     }
 
     @JsonProperty("subscriptions")
@@ -71,7 +80,9 @@ public Map<String, NonPersistentReplicatorStats> getNonPersistentReplicators() {
     }
 
     /** List of connected publishers on this non-persistent topic w/ their stats. */
-    public List<? extends NonPersistentPublisherStats> nonPersistentPublishers;
+    private List<NonPersistentPublisherStats> nonPersistentPublishers;
+
+    private Map<String, NonPersistentPublisherStats> nonPersistentPublishersMap;
 
     /** Map of non-persistent subscriptions with their individual statistics. */
     public Map<String, ? extends NonPersistentSubscriptionStats> nonPersistentSubscriptions;
@@ -81,7 +92,25 @@ public Map<String, NonPersistentReplicatorStats> getNonPersistentReplicators() {
 
     @SuppressFBWarnings(value = "MF_CLASS_MASKS_FIELD", justification = "expected to override")
     public List<NonPersistentPublisherStats> getPublishers() {
-        return (List<NonPersistentPublisherStats>) nonPersistentPublishers;
+        return Stream.concat(nonPersistentPublishers.stream()
+                                .sorted(Comparator.comparing(NonPersistentPublisherStats::getProducerName)),
+                nonPersistentPublishersMap.values()
+                        .stream().sorted(Comparator.comparing(NonPersistentPublisherStats::getProducerName)))
+                .collect(Collectors.toList());
+    }
+
+    public void setPublishers(List<? extends PublisherStats> statsList) {
+        this.nonPersistentPublishers.clear();
+        this.nonPersistentPublishersMap.clear();
+        statsList.forEach(s -> addPublisher((NonPersistentPublisherStatsImpl) s));
+    }
+
+    public void addPublisher(NonPersistentPublisherStatsImpl stats) {
+        if (stats.isSupportsPartialProducer()) {
+            nonPersistentPublishersMap.put(stats.getProducerName(), stats);
+        } else {
+            nonPersistentPublishers.add(stats);
+        }
     }
 
     @SuppressFBWarnings(value = "MF_CLASS_MASKS_FIELD", justification = "expected to override")
@@ -101,6 +130,7 @@ public double getMsgDropRate() {
 
     public NonPersistentTopicStatsImpl() {
         this.nonPersistentPublishers = new ArrayList<>();
+        this.nonPersistentPublishersMap = new ConcurrentHashMap<>();
         this.nonPersistentSubscriptions = new HashMap<>();
         this.nonPersistentReplicators = new TreeMap<>();
     }
@@ -108,6 +138,7 @@ public NonPersistentTopicStatsImpl() {
     public void reset() {
         super.reset();
         this.nonPersistentPublishers.clear();
+        this.nonPersistentPublishersMap.clear();
         this.nonPersistentSubscriptions.clear();
         this.nonPersistentReplicators.clear();
         this.msgDropRate = 0;
@@ -121,18 +152,30 @@ public NonPersistentTopicStatsImpl add(NonPersistentTopicStats ts) {
         super.add(stats);
         this.msgDropRate += stats.msgDropRate;
 
-        if (this.getNonPersistentPublishers().size() != stats.getNonPersistentPublishers().size()) {
-            for (int i = 0; i < stats.getNonPersistentPublishers().size(); i++) {
-                NonPersistentPublisherStatsImpl publisherStats = new NonPersistentPublisherStatsImpl();
-                this.getNonPersistentPublishers().add(publisherStats
-                        .add((NonPersistentPublisherStatsImpl) stats.getNonPersistentPublishers().get(i)));
-            }
-        } else {
-            for (int i = 0; i < stats.getNonPersistentPublishers().size(); i++) {
-                ((NonPersistentPublisherStatsImpl) this.getNonPersistentPublishers().get(i))
-                        .add((NonPersistentPublisherStatsImpl) stats.getNonPersistentPublishers().get(i));
+        stats.getNonPersistentPublishers().forEach(s -> {
+            if (s.isSupportsPartialProducer()) {
+                ((NonPersistentPublisherStatsImpl) this.nonPersistentPublishersMap
+                        .computeIfAbsent(s.getProducerName(), key -> {
+                            final NonPersistentPublisherStatsImpl newStats = new NonPersistentPublisherStatsImpl();
+                            newStats.setSupportsPartialProducer(true);
+                            newStats.setProducerName(s.getProducerName());
+                            return newStats;
+                        })).add((NonPersistentPublisherStatsImpl) s);
+            } else {
+                if (this.nonPersistentPublishers.size() != stats.getNonPersistentPublishers().size()) {
+                    for (int i = 0; i < stats.getNonPersistentPublishers().size(); i++) {
+                        NonPersistentPublisherStatsImpl newStats = new NonPersistentPublisherStatsImpl();
+                        newStats.setSupportsPartialProducer(false);
+                        this.nonPersistentPublishers.add(newStats.add((NonPersistentPublisherStatsImpl) s));
+                    }
+                } else {
+                    for (int i = 0; i < stats.getNonPersistentPublishers().size(); i++) {
+                        ((NonPersistentPublisherStatsImpl) this.nonPersistentPublishers.get(i))
+                                .add((NonPersistentPublisherStatsImpl) s);
+                    }
+                }
             }
-        }
+        });
 
         if (this.getNonPersistentSubscriptions().size() != stats.getNonPersistentSubscriptions().size()) {
             for (String subscription : stats.getNonPersistentSubscriptions().keySet()) {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java
index 41de0a33ef8bd..93d42e6f07842 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/PublisherStatsImpl.java
@@ -49,6 +49,9 @@ public class PublisherStatsImpl implements PublisherStats {
     /** Id of this publisher. */
     public long producerId;
 
+    /** Whether partial producer is supported at client. */
+    public boolean supportsPartialProducer;
+
     /** Producer name. */
     @JsonIgnore
     private int producerNameOffset = -1;
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java
index 0aa7c971820bb..1164dd35b29a7 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java
@@ -20,13 +20,18 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import lombok.AccessLevel;
 import lombok.Data;
 import lombok.Getter;
+import lombok.Setter;
 import org.apache.pulsar.common.policies.data.PublisherStats;
 import org.apache.pulsar.common.policies.data.ReplicatorStats;
 import org.apache.pulsar.common.policies.data.SubscriptionStats;
@@ -93,7 +98,12 @@ public class TopicStatsImpl implements TopicStats {
 
     /** List of connected publishers on this topic w/ their stats. */
     @Getter(AccessLevel.NONE)
-    public List<PublisherStatsImpl> publishers;
+    @Setter(AccessLevel.NONE)
+    private List<PublisherStatsImpl> publishers;
+
+    @Getter(AccessLevel.NONE)
+    @Setter(AccessLevel.NONE)
+    private Map<String, PublisherStatsImpl> publishersMap;
 
     public int waitingPublishers;
 
@@ -120,7 +130,23 @@ public class TopicStatsImpl implements TopicStats {
     public CompactionStatsImpl compaction;
 
     public List<? extends PublisherStats> getPublishers() {
-        return publishers;
+        return Stream.concat(publishers.stream().sorted(Comparator.comparing(PublisherStatsImpl::getProducerName)),
+                publishersMap.values().stream().sorted(Comparator.comparing(PublisherStatsImpl::getProducerName)))
+                .collect(Collectors.toList());
+    }
+
+    public void setPublishers(List<? extends PublisherStats> statsList) {
+        this.publishers.clear();
+        this.publishersMap.clear();
+        statsList.forEach(s -> addPublisher((PublisherStatsImpl) s));
+    }
+
+    public void addPublisher(PublisherStatsImpl stats) {
+        if (stats.isSupportsPartialProducer()) {
+            publishersMap.put(stats.getProducerName(), stats);
+        } else {
+            publishers.add(stats);
+        }
     }
 
     public Map<String, ? extends SubscriptionStats> getSubscriptions() {
@@ -133,6 +159,7 @@ public List<? extends PublisherStats> getPublishers() {
 
     public TopicStatsImpl() {
         this.publishers = new ArrayList<>();
+        this.publishersMap = new ConcurrentHashMap<>();
         this.subscriptions = new HashMap<>();
         this.replication = new TreeMap<>();
         this.compaction = new CompactionStatsImpl();
@@ -152,6 +179,7 @@ public void reset() {
         this.bytesOutCounter = 0;
         this.msgOutCounter = 0;
         this.publishers.clear();
+        this.publishersMap.clear();
         this.subscriptions.clear();
         this.waitingPublishers = 0;
         this.replication.clear();
@@ -188,16 +216,30 @@ public TopicStatsImpl add(TopicStats ts) {
         this.offloadedStorageSize += stats.offloadedStorageSize;
         this.nonContiguousDeletedMessagesRanges += stats.nonContiguousDeletedMessagesRanges;
         this.nonContiguousDeletedMessagesRangesSerializedSize += stats.nonContiguousDeletedMessagesRangesSerializedSize;
-        if (this.publishers.size() != stats.publishers.size()) {
-            for (int i = 0; i < stats.publishers.size(); i++) {
-                PublisherStatsImpl publisherStats = new PublisherStatsImpl();
-                this.publishers.add(publisherStats.add(stats.publishers.get(i)));
-            }
-        } else {
-            for (int i = 0; i < stats.publishers.size(); i++) {
-                this.publishers.get(i).add(stats.publishers.get(i));
-            }
-        }
+
+        stats.getPublishers().forEach(s -> {
+           if (s.isSupportsPartialProducer()) {
+               this.publishersMap.computeIfAbsent(s.getProducerName(), key -> {
+                   final PublisherStatsImpl newStats = new PublisherStatsImpl();
+                   newStats.setSupportsPartialProducer(true);
+                   newStats.setProducerName(s.getProducerName());
+                   return newStats;
+               }).add((PublisherStatsImpl) s);
+           } else {
+               if (this.publishers.size() != stats.publishers.size()) {
+                   for (int i = 0; i < stats.publishers.size(); i++) {
+                       PublisherStatsImpl newStats = new PublisherStatsImpl();
+                       newStats.setSupportsPartialProducer(false);
+                       this.publishers.add(newStats.add(stats.publishers.get(i)));
+                   }
+               } else {
+                   for (int i = 0; i < stats.publishers.size(); i++) {
+                       this.publishers.get(i).add(stats.publishers.get(i));
+                   }
+               }
+           }
+        });
+
         if (this.subscriptions.size() != stats.subscriptions.size()) {
             for (String subscription : stats.subscriptions.keySet()) {
                 SubscriptionStatsImpl subscriptionStats = new SubscriptionStatsImpl();
@@ -230,5 +272,4 @@ public TopicStatsImpl add(TopicStats ts) {
         }
         return this;
     }
-
 }
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index 93a47f0cecf63..3a130aeae4d0e 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -180,6 +180,7 @@ public static ByteBuf newConnect(String authMethodName, String authData, String
     private static void setFeatureFlags(FeatureFlags flags) {
         flags.setSupportsAuthRefresh(true);
         flags.setSupportsBrokerEntryMetadata(true);
+        flags.setSupportsPartialProducer(true);
     }
 
     public static ByteBuf newConnect(String authMethodName, String authData, int protocolVersion, String libVersion,
diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto
index b6e80ac620124..12367a2236082 100644
--- a/pulsar-common/src/main/proto/PulsarApi.proto
+++ b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -292,6 +292,7 @@ message CommandConnect {
 message FeatureFlags {
   optional bool supports_auth_refresh = 1 [default = false];
   optional bool supports_broker_entry_metadata = 2 [default = false];
+  optional bool supports_partial_producer = 3 [default = false];
 }
 
 message CommandConnected {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PartitionedTopicStatsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PartitionedTopicStatsTest.java
index 8d9627bf0e21e..aa2b90ba8a96d 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PartitionedTopicStatsTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PartitionedTopicStatsTest.java
@@ -37,7 +37,7 @@ public void testPartitionedTopicStats() {
         partitionedTopicStats.msgThroughputOut = 1;
         partitionedTopicStats.averageMsgSize = 1;
         partitionedTopicStats.storageSize = 1;
-        partitionedTopicStats.publishers.add(new PublisherStatsImpl());
+        partitionedTopicStats.addPublisher((new PublisherStatsImpl()));
         partitionedTopicStats.subscriptions.put("test_ns", new SubscriptionStatsImpl());
         partitionedTopicStats.replication.put("test_ns", new ReplicatorStatsImpl());
         partitionedTopicStats.metadata.partitions = 1;
@@ -49,7 +49,7 @@ public void testPartitionedTopicStats() {
         assertEquals(partitionedTopicStats.msgThroughputOut, 0.0);
         assertEquals(partitionedTopicStats.averageMsgSize, 0.0);
         assertEquals(partitionedTopicStats.storageSize, 0);
-        assertEquals(partitionedTopicStats.publishers.size(), 0);
+        assertEquals(partitionedTopicStats.getPublishers().size(), 0);
         assertEquals(partitionedTopicStats.subscriptions.size(), 0);
         assertEquals(partitionedTopicStats.replication.size(), 0);
         assertEquals(partitionedTopicStats.metadata.partitions, 0);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PersistentTopicStatsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PersistentTopicStatsTest.java
index 4a82651467137..0eb1997ffd29c 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PersistentTopicStatsTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PersistentTopicStatsTest.java
@@ -25,6 +25,9 @@
 import org.apache.pulsar.common.policies.data.stats.SubscriptionStatsImpl;
 import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
 import org.testng.annotations.Test;
+import org.testng.collections.Maps;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 public class PersistentTopicStatsTest {
 
@@ -38,7 +41,7 @@ public void testPersistentTopicStats() {
         topicStats.averageMsgSize = 1;
         topicStats.storageSize = 1;
         topicStats.offloadedStorageSize = 1;
-        topicStats.publishers.add(new PublisherStatsImpl());
+        topicStats.addPublisher(new PublisherStatsImpl());
         topicStats.subscriptions.put("test_ns", new SubscriptionStatsImpl());
         topicStats.replication.put("test_ns", new ReplicatorStatsImpl());
         TopicStatsImpl target = new TopicStatsImpl();
@@ -53,7 +56,7 @@ public void testPersistentTopicStats() {
         assertEquals(topicStats.lastOffloadSuccessTimeStamp, 0);
         assertEquals(topicStats.lastOffloadFailureTimeStamp, 0);
         assertEquals(topicStats.storageSize, 1);
-        assertEquals(topicStats.publishers.size(), 1);
+        assertEquals(topicStats.getPublishers().size(), 1);
         assertEquals(topicStats.subscriptions.size(), 1);
         assertEquals(topicStats.replication.size(), 1);
         assertEquals(topicStats.compaction.lastCompactionRemovedEventCount, 0);
@@ -67,14 +70,14 @@ public void testPersistentTopicStats() {
         assertEquals(topicStats.msgThroughputOut, 0.0);
         assertEquals(topicStats.averageMsgSize, 0.0);
         assertEquals(topicStats.storageSize, 0);
-        assertEquals(topicStats.publishers.size(), 0);
+        assertEquals(topicStats.getPublishers().size(), 0);
         assertEquals(topicStats.subscriptions.size(), 0);
         assertEquals(topicStats.replication.size(), 0);
         assertEquals(topicStats.offloadedStorageSize, 0);
     }
 
     @Test
-    public void testPersistentTopicStatsAggregation() {
+    public void testPersistentTopicStatsAggregationPartialProducerIsNotSupported() {
         TopicStatsImpl topicStats1 = new TopicStatsImpl();
         topicStats1.msgRateIn = 1;
         topicStats1.msgThroughputIn = 1;
@@ -82,7 +85,10 @@ public void testPersistentTopicStatsAggregation() {
         topicStats1.msgThroughputOut = 1;
         topicStats1.averageMsgSize = 1;
         topicStats1.storageSize = 1;
-        topicStats1.publishers.add(new PublisherStatsImpl());
+        final PublisherStatsImpl publisherStats1 = new PublisherStatsImpl();
+        publisherStats1.setSupportsPartialProducer(false);
+        publisherStats1.setProducerName("name1");
+        topicStats1.addPublisher(publisherStats1);
         topicStats1.subscriptions.put("test_ns", new SubscriptionStatsImpl());
         topicStats1.replication.put("test_ns", new ReplicatorStatsImpl());
 
@@ -93,7 +99,10 @@ public void testPersistentTopicStatsAggregation() {
         topicStats2.msgThroughputOut = 4;
         topicStats2.averageMsgSize = 5;
         topicStats2.storageSize = 6;
-        topicStats2.publishers.add(new PublisherStatsImpl());
+        final PublisherStatsImpl publisherStats2 = new PublisherStatsImpl();
+        publisherStats2.setSupportsPartialProducer(false);
+        publisherStats2.setProducerName("name1");
+        topicStats2.addPublisher(publisherStats2);
         topicStats2.subscriptions.put("test_ns", new SubscriptionStatsImpl());
         topicStats2.replication.put("test_ns", new ReplicatorStatsImpl());
 
@@ -107,7 +116,121 @@ public void testPersistentTopicStatsAggregation() {
         assertEquals(target.msgThroughputOut, 5.0);
         assertEquals(target.averageMsgSize, 3.0);
         assertEquals(target.storageSize, 7);
-        assertEquals(target.publishers.size(), 1);
+        assertEquals(target.getPublishers().size(), 1);
+        assertEquals(target.subscriptions.size(), 1);
+        assertEquals(target.replication.size(), 1);
+    }
+
+    @Test
+    public void testPersistentTopicStatsAggregationPartialProducerSupported() {
+        TopicStatsImpl topicStats1 = new TopicStatsImpl();
+        topicStats1.msgRateIn = 1;
+        topicStats1.msgThroughputIn = 1;
+        topicStats1.msgRateOut = 1;
+        topicStats1.msgThroughputOut = 1;
+        topicStats1.averageMsgSize = 1;
+        topicStats1.storageSize = 1;
+        final PublisherStatsImpl publisherStats1 = new PublisherStatsImpl();
+        publisherStats1.setSupportsPartialProducer(true);
+        publisherStats1.setProducerName("name1");
+        topicStats1.addPublisher(publisherStats1);
+        topicStats1.subscriptions.put("test_ns", new SubscriptionStatsImpl());
+        topicStats1.replication.put("test_ns", new ReplicatorStatsImpl());
+
+        TopicStatsImpl topicStats2 = new TopicStatsImpl();
+        topicStats2.msgRateIn = 1;
+        topicStats2.msgThroughputIn = 2;
+        topicStats2.msgRateOut = 3;
+        topicStats2.msgThroughputOut = 4;
+        topicStats2.averageMsgSize = 5;
+        topicStats2.storageSize = 6;
+        final PublisherStatsImpl publisherStats2 = new PublisherStatsImpl();
+        publisherStats2.setSupportsPartialProducer(true);
+        publisherStats2.setProducerName("name1");
+        topicStats2.addPublisher(publisherStats2);
+        topicStats2.subscriptions.put("test_ns", new SubscriptionStatsImpl());
+        topicStats2.replication.put("test_ns", new ReplicatorStatsImpl());
+
+        TopicStatsImpl target = new TopicStatsImpl();
+        target.add(topicStats1);
+        target.add(topicStats2);
+
+        assertEquals(target.msgRateIn, 2.0);
+        assertEquals(target.msgThroughputIn, 3.0);
+        assertEquals(target.msgRateOut, 4.0);
+        assertEquals(target.msgThroughputOut, 5.0);
+        assertEquals(target.averageMsgSize, 3.0);
+        assertEquals(target.storageSize, 7);
+        assertEquals(target.getPublishers().size(), 1);
+        assertEquals(target.getPublishers().get(0).getProducerName(), "name1");
+        assertEquals(target.subscriptions.size(), 1);
+        assertEquals(target.replication.size(), 1);
+    }
+
+    @Test
+    public void testPersistentTopicStatsAggregationByProducerName() {
+        TopicStatsImpl topicStats1 = new TopicStatsImpl();
+        topicStats1.msgRateIn = 1;
+        topicStats1.msgThroughputIn = 1;
+        topicStats1.msgRateOut = 1;
+        topicStats1.msgThroughputOut = 1;
+        topicStats1.averageMsgSize = 1;
+        topicStats1.storageSize = 1;
+        final PublisherStatsImpl publisherStats1 = new PublisherStatsImpl();
+        publisherStats1.setSupportsPartialProducer(true);
+        publisherStats1.msgRateIn = 1;
+        publisherStats1.setProducerName("name1");
+        topicStats1.addPublisher(publisherStats1);
+        topicStats1.subscriptions.put("test_ns", new SubscriptionStatsImpl());
+        topicStats1.replication.put("test_ns", new ReplicatorStatsImpl());
+
+        TopicStatsImpl topicStats2 = new TopicStatsImpl();
+        topicStats2.msgRateIn = 1;
+        topicStats2.msgThroughputIn = 2;
+        topicStats2.msgRateOut = 3;
+        topicStats2.msgThroughputOut = 4;
+        topicStats2.averageMsgSize = 5;
+        topicStats2.storageSize = 6;
+        final PublisherStatsImpl publisherStats2 = new PublisherStatsImpl();
+        publisherStats2.setSupportsPartialProducer(true);
+        publisherStats2.msgRateIn = 1;
+        publisherStats2.setProducerName("name1");
+        topicStats2.addPublisher(publisherStats2);
+        topicStats2.subscriptions.put("test_ns", new SubscriptionStatsImpl());
+        topicStats2.replication.put("test_ns", new ReplicatorStatsImpl());
+
+        TopicStatsImpl topicStats3 = new TopicStatsImpl();
+        topicStats3.msgRateIn = 0;
+        topicStats3.msgThroughputIn = 0;
+        topicStats3.msgRateOut = 0;
+        topicStats3.msgThroughputOut = 0;
+        topicStats3.averageMsgSize = 0;
+        topicStats3.storageSize = 0;
+        final PublisherStatsImpl publisherStats3 = new PublisherStatsImpl();
+        publisherStats3.setSupportsPartialProducer(true);
+        publisherStats3.msgRateIn = 1;
+        publisherStats3.setProducerName("name2");
+        topicStats3.addPublisher(publisherStats3);
+        topicStats3.subscriptions.put("test_ns", new SubscriptionStatsImpl());
+        topicStats3.replication.put("test_ns", new ReplicatorStatsImpl());
+
+        TopicStatsImpl target = new TopicStatsImpl();
+        target.add(topicStats1);
+        target.add(topicStats2);
+        target.add(topicStats3);
+
+        assertEquals(target.msgRateIn, 2.0);
+        assertEquals(target.msgThroughputIn, 3.0);
+        assertEquals(target.msgRateOut, 4.0);
+        assertEquals(target.msgThroughputOut, 5.0);
+        assertEquals(target.averageMsgSize, 2.0);
+        assertEquals(target.storageSize, 7);
+        assertEquals(target.getPublishers().size(), 2);
+        final Map<String, Double> expectedPublishersMap = Maps.newHashMap();
+        expectedPublishersMap.put("name1", 2.0);
+        expectedPublishersMap.put("name2", 1.0);
+        assertEquals(target.getPublishers().stream().collect(
+                Collectors.toMap(PublisherStats::getProducerName, e -> ((PublisherStatsImpl) e).msgRateIn)), expectedPublishersMap);
         assertEquals(target.subscriptions.size(), 1);
         assertEquals(target.replication.size(), 1);
     }
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PublisherStatsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PublisherStatsTest.java
index 9f34a9128e48f..2aa1974bd659a 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PublisherStatsTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PublisherStatsTest.java
@@ -46,7 +46,8 @@ public void testPublisherStats() throws Exception {
             "address",
             "connectedSince",
             "clientVersion",
-            "producerName"
+            "producerName",
+            "supportsPartialProducer"
         );
 
         PublisherStatsImpl stats = new PublisherStatsImpl();