codelipenghui commented on code in PR #15867:
URL: https://github.com/apache/pulsar/pull/15867#discussion_r898013253
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java:
##########
@@ -345,6 +346,9 @@ private static void
printNamespaceStats(SimpleTextOutputStream stream, String cl
metric(stream, cluster, namespace, "pulsar_subscription_delayed",
stats.msgDelayed);
+ metric(stream, cluster, namespace,
"pulsar_delayed_delivery_tracker_memory_usage",
Review Comment:
I think `pulsar_delayed_message_index_size_bytes` is better? The usage is a
little confusing which can mean *%
And delayed_delivery_tracker is just the name of our implementation. From
the user's perspective, the delayed message index is more easy to understand
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -308,4 +307,70 @@ public void
testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exce
assertEquals(admin.topics().getList(ns).size(), 0);
pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned");
}
+
+
+ @Test
+ public void testNamespaceLevelDelayedDeliveryTrackerMemoryUsageMetric()
throws Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, false);
+ }
+
+ @Test
+ public void testTopicLevelDelayedDeliveryTrackerMemoryUsageMetric() throws
Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, true);
+ }
+
+
+ private void testDelayedDeliveryTrackerMemoryUsageMetric(String topic,
boolean exposeTopicLevelMetrics) throws Exception {
+ PulsarClient client = pulsar.getClient();
+ String namespace = TopicName.get(topic).getNamespace();
+ admin.namespaces().createNamespace(namespace);
+
+ @Cleanup
+ Producer<String> producer =
client.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ @Cleanup
+ Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test_sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .messageListener((MessageListener<String>) (consumer1, msg) ->
{
+ try {
+ consumer1.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ e.printStackTrace();
+ }
+ })
+ .subscribe();
+ for (int a = 0; a < 100; a++) {
+
producer.newMessage().value(UUID.randomUUID().toString()).deliverAfter(30,
TimeUnit.SECONDS).send();
+ }
+ producer.flush();
+
+ //wait for Dispatcher build delayed message index.
+ for (int a = 0; a < 100; a++) {
+ Thread.sleep(1);
+ }
+
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
Review Comment:
Close after the test. Or add try resource.
##########
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java:
##########
@@ -131,6 +131,9 @@ public class TopicStatsImpl implements TopicStats {
/** The serialized size of non-contiguous deleted messages ranges. */
public int nonContiguousDeletedMessagesRangesSerializedSize;
+ /** The size of InMemoryDelayedDeliveryTracer memory usage. */
+ public int delayedTrackerMemoryUsage;
Review Comment:
```suggestion
public int delayedMessageIndexSizeInBytes;
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -308,4 +307,70 @@ public void
testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exce
assertEquals(admin.topics().getList(ns).size(), 0);
pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned");
}
+
+
+ @Test
+ public void testNamespaceLevelDelayedDeliveryTrackerMemoryUsageMetric()
throws Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, false);
+ }
+
+ @Test
+ public void testTopicLevelDelayedDeliveryTrackerMemoryUsageMetric() throws
Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, true);
+ }
Review Comment:
Can replace with a data provider.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -308,4 +307,70 @@ public void
testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exce
assertEquals(admin.topics().getList(ns).size(), 0);
pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned");
}
+
+
+ @Test
+ public void testNamespaceLevelDelayedDeliveryTrackerMemoryUsageMetric()
throws Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, false);
+ }
+
+ @Test
+ public void testTopicLevelDelayedDeliveryTrackerMemoryUsageMetric() throws
Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, true);
+ }
+
+
+ private void testDelayedDeliveryTrackerMemoryUsageMetric(String topic,
boolean exposeTopicLevelMetrics) throws Exception {
+ PulsarClient client = pulsar.getClient();
+ String namespace = TopicName.get(topic).getNamespace();
+ admin.namespaces().createNamespace(namespace);
+
+ @Cleanup
+ Producer<String> producer =
client.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ @Cleanup
+ Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test_sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .messageListener((MessageListener<String>) (consumer1, msg) ->
{
+ try {
+ consumer1.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ e.printStackTrace();
+ }
+ })
+ .subscribe();
+ for (int a = 0; a < 100; a++) {
+
producer.newMessage().value(UUID.randomUUID().toString()).deliverAfter(30,
TimeUnit.SECONDS).send();
+ }
+ producer.flush();
+
+ //wait for Dispatcher build delayed message index.
+ for (int a = 0; a < 100; a++) {
+ Thread.sleep(1);
+ }
Review Comment:
Here might introduce a flaky test because the delayed message index might
not complete within 100ms
It's better to use Awaitibility to check some state for the delayed message
tracker to make sure we are able to continue the test.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -308,4 +307,70 @@ public void
testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exce
assertEquals(admin.topics().getList(ns).size(), 0);
pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned");
}
+
+
+ @Test
+ public void testNamespaceLevelDelayedDeliveryTrackerMemoryUsageMetric()
throws Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, false);
+ }
+
+ @Test
+ public void testTopicLevelDelayedDeliveryTrackerMemoryUsageMetric() throws
Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, true);
+ }
+
+
+ private void testDelayedDeliveryTrackerMemoryUsageMetric(String topic,
boolean exposeTopicLevelMetrics) throws Exception {
+ PulsarClient client = pulsar.getClient();
+ String namespace = TopicName.get(topic).getNamespace();
+ admin.namespaces().createNamespace(namespace);
+
+ @Cleanup
+ Producer<String> producer =
client.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ @Cleanup
+ Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test_sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .messageListener((MessageListener<String>) (consumer1, msg) ->
{
+ try {
+ consumer1.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ e.printStackTrace();
+ }
+ })
+ .subscribe();
+ for (int a = 0; a < 100; a++) {
+
producer.newMessage().value(UUID.randomUUID().toString()).deliverAfter(30,
TimeUnit.SECONDS).send();
+ }
+ producer.flush();
+
+ //wait for Dispatcher build delayed message index.
+ for (int a = 0; a < 100; a++) {
+ Thread.sleep(1);
+ }
+
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ PrometheusMetricsGenerator.generate(pulsar, exposeTopicLevelMetrics,
true, true, output);
+ String metricsStr = output.toString(StandardCharsets.UTF_8);
+
+ Multimap<String, PrometheusMetricsTest.Metric> metricsMap =
PrometheusMetricsTest.parseMetrics(metricsStr);
+ Collection<PrometheusMetricsTest.Metric> metrics =
metricsMap.get("pulsar_delayed_delivery_tracker_memory_usage");
+ Assert.assertTrue(metrics.size() > 0);
+
+ int num = 0;
+ for (PrometheusMetricsTest.Metric metric : metrics) {
+ if (exposeTopicLevelMetrics &&
metric.tags.get("topic").equals(topic)) {
+ Assert.assertTrue(metric.value > 0);
Review Comment:
Need to check not namespace level metrics.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -308,4 +307,70 @@ public void
testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exce
assertEquals(admin.topics().getList(ns).size(), 0);
pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned");
}
+
+
+ @Test
+ public void testNamespaceLevelDelayedDeliveryTrackerMemoryUsageMetric()
throws Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, false);
+ }
+
+ @Test
+ public void testTopicLevelDelayedDeliveryTrackerMemoryUsageMetric() throws
Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, true);
+ }
+
+
+ private void testDelayedDeliveryTrackerMemoryUsageMetric(String topic,
boolean exposeTopicLevelMetrics) throws Exception {
+ PulsarClient client = pulsar.getClient();
+ String namespace = TopicName.get(topic).getNamespace();
+ admin.namespaces().createNamespace(namespace);
+
+ @Cleanup
+ Producer<String> producer =
client.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ @Cleanup
+ Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test_sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .messageListener((MessageListener<String>) (consumer1, msg) ->
{
+ try {
+ consumer1.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ e.printStackTrace();
+ }
+ })
+ .subscribe();
+ for (int a = 0; a < 100; a++) {
+
producer.newMessage().value(UUID.randomUUID().toString()).deliverAfter(30,
TimeUnit.SECONDS).send();
Review Comment:
```suggestion
producer.newMessage().value(UUID.randomUUID().toString()).deliverAfter(30,
TimeUnit.SECONDS).sendAsync();
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -308,4 +307,70 @@ public void
testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exce
assertEquals(admin.topics().getList(ns).size(), 0);
pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned");
}
+
+
+ @Test
+ public void testNamespaceLevelDelayedDeliveryTrackerMemoryUsageMetric()
throws Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, false);
+ }
+
+ @Test
+ public void testTopicLevelDelayedDeliveryTrackerMemoryUsageMetric() throws
Exception {
+ String topic = "persistent://prop/autoNs/test_delayed_message_metric"
+ UUID.randomUUID();
+ testDelayedDeliveryTrackerMemoryUsageMetric(topic, true);
+ }
+
+
+ private void testDelayedDeliveryTrackerMemoryUsageMetric(String topic,
boolean exposeTopicLevelMetrics) throws Exception {
+ PulsarClient client = pulsar.getClient();
+ String namespace = TopicName.get(topic).getNamespace();
+ admin.namespaces().createNamespace(namespace);
+
+ @Cleanup
+ Producer<String> producer =
client.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ @Cleanup
+ Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test_sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .messageListener((MessageListener<String>) (consumer1, msg) ->
{
+ try {
+ consumer1.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ e.printStackTrace();
+ }
+ })
+ .subscribe();
+ for (int a = 0; a < 100; a++) {
+
producer.newMessage().value(UUID.randomUUID().toString()).deliverAfter(30,
TimeUnit.SECONDS).send();
+ }
+ producer.flush();
+
+ //wait for Dispatcher build delayed message index.
+ for (int a = 0; a < 100; a++) {
+ Thread.sleep(1);
+ }
+
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ PrometheusMetricsGenerator.generate(pulsar, exposeTopicLevelMetrics,
true, true, output);
+ String metricsStr = output.toString(StandardCharsets.UTF_8);
+
+ Multimap<String, PrometheusMetricsTest.Metric> metricsMap =
PrometheusMetricsTest.parseMetrics(metricsStr);
+ Collection<PrometheusMetricsTest.Metric> metrics =
metricsMap.get("pulsar_delayed_delivery_tracker_memory_usage");
+ Assert.assertTrue(metrics.size() > 0);
+
+ int num = 0;
+ for (PrometheusMetricsTest.Metric metric : metrics) {
+ if (exposeTopicLevelMetrics &&
metric.tags.get("topic").equals(topic)) {
+ Assert.assertTrue(metric.value > 0);
+ num++;
+ } else if (!exposeTopicLevelMetrics &&
metric.tags.get("namespace").equals(namespace)) {
+ Assert.assertTrue(metric.value > 0);
+ num++;
Review Comment:
Need to check no topic level metrics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
