You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/06/02 08:58:00 UTC
(pulsar) 04/05: [fix][broker] expose consumer name for partitioned topic stats (#24360)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c9d23a0ae27a5e4368274e6eb6a2785ef7a1a514
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Mon Jun 2 03:50:24 2025 +0800
[fix][broker] expose consumer name for partitioned topic stats (#24360)
(cherry picked from commit 1c308710840e1f15d4421d429202fc6d212bf31a)
---
.../apache/pulsar/broker/admin/AdminApi2Test.java | 23 ++++++++++++++++++++++
.../policies/data/stats/ConsumerStatsImpl.java | 1 +
2 files changed, 24 insertions(+)
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 cc8423a7ac8..5d54eadb0b2 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
@@ -1865,6 +1865,29 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
+
+ @Test
+ public void testPartitionedTopicStatsIncludeConsumerName() throws PulsarClientException, PulsarAdminException {
+ final String topic = "persistent://" + defaultNamespace + "/" + UUID.randomUUID();
+ admin.topics().createPartitionedTopic(topic, 2);
+ final String subName = "sub-name";
+ final String consumerName = "consumer-name";
+
+ @Cleanup
+ final PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getWebServiceAddress()).build();
+
+ @Cleanup
+ final Consumer<byte[]> consumer = client.newConsumer()
+ .topic(topic)
+ .subscriptionName(subName)
+ .consumerName(consumerName)
+ .subscribe();
+
+ final TopicStats topicStats = admin.topics().getPartitionedStats(topic, false);
+
+ assertEquals(topicStats.getSubscriptions().get(subName).getConsumers().get(0).getConsumerName(), consumerName);
+ }
+
@Test
public void testPreciseBacklogForPartitionedTopic() throws PulsarClientException, PulsarAdminException {
final String topic = "persistent://" + defaultNamespace + "/precise-back-log-for-partitioned-topic";
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java
index ddae2e71356..a38638928b3 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java
@@ -130,6 +130,7 @@ public class ConsumerStatsImpl implements ConsumerStats {
this.unackedMessages += stats.unackedMessages;
this.blockedConsumerOnUnackedMsgs = stats.blockedConsumerOnUnackedMsgs;
this.readPositionWhenJoining = stats.readPositionWhenJoining;
+ this.consumerName = stats.consumerName;
return this;
}