You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2024/05/02 15:27:34 UTC
(pulsar) branch branch-2.11 updated (305aaca151a -> a8cd7befe6c)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 305aaca151a [fix][test] Fix the flaky tests of ManagedLedgerImplUtilsTest (#22611)
new a03543a4e48 [improve][broker] Add `topic_load_failed` metric (#19236)
new 58409c911dd [fix] [broker] Fix metrics pulsar_topic_load_failed_count is 0 when load non-persistent topic fails and fix the flaky test testBrokerStatsTopicLoadFailed (#22580)
new a8cd7befe6c [cleanup] [test] Cleanup unnecessary imports
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pulsar/broker/service/BrokerService.java | 12 +-
.../apache/pulsar/broker/service/PulsarStats.java | 4 +
.../broker/stats/BrokerOperabilityMetrics.java | 10 +-
.../pulsar/broker/service/BrokerServiceTest.java | 127 +++++++++++++++++++++
.../pulsar/broker/stats/PrometheusMetricsTest.java | 4 +
5 files changed, 155 insertions(+), 2 deletions(-)
(pulsar) 02/03: [fix] [broker] Fix metrics pulsar_topic_load_failed_count is 0 when load non-persistent topic fails and fix the flaky test testBrokerStatsTopicLoadFailed (#22580)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 58409c911dd55173adc7f3524f4ae26abb0ca6a3
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Apr 29 20:46:43 2024 +0800
[fix] [broker] Fix metrics pulsar_topic_load_failed_count is 0 when load non-persistent topic fails and fix the flaky test testBrokerStatsTopicLoadFailed (#22580)
(cherry picked from commit 340d60df0be32ed26586f292a8d24a8a6663aba2)
---
.../pulsar/broker/service/BrokerService.java | 3 +-
.../pulsar/broker/service/BrokerServiceTest.java | 186 +++++++++++++--------
2 files changed, 121 insertions(+), 68 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 65366f7eb22..f05463a128c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1200,7 +1200,8 @@ public class BrokerService implements Closeable {
nonPersistentTopic = newTopic(topic, null, this, NonPersistentTopic.class);
} catch (Exception e) {
log.warn("Failed to create topic {}", topic, e);
- return FutureUtil.failedFuture(e);
+ topicFuture.completeExceptionally(e);
+ return topicFuture;
}
CompletableFuture<Void> isOwner = checkTopicNsOwnership(topic);
isOwner.thenRun(() -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index ea10b24b385..70919716df7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -20,9 +20,12 @@ package org.apache.pulsar.broker.service;
import static org.apache.pulsar.common.naming.SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN;
import static org.apache.pulsar.common.naming.SystemTopicNames.TRANSACTION_COORDINATOR_LOG;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
@@ -30,11 +33,11 @@ import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
+import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
@@ -42,9 +45,12 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
+import java.io.Reader;
+import java.io.StringReader;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -71,6 +77,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
@@ -80,6 +87,7 @@ import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient;
import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
import org.apache.pulsar.client.admin.BrokerStats;
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -110,7 +118,11 @@ import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.apache.pulsar.compaction.Compactor;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.MockZooKeeper;
import org.awaitility.Awaitility;
+import org.glassfish.jersey.client.JerseyClient;
+import org.glassfish.jersey.client.JerseyClientBuilder;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -1630,81 +1642,121 @@ public class BrokerServiceTest extends BrokerTestBase {
});
}
-
@Test
- public void testBrokerStatsTopicLoadFailed() throws Exception {
- admin.namespaces().createNamespace("prop/ns-test");
-
- String persistentTopic = "persistent://prop/ns-test/topic1_" + UUID.randomUUID();
- String nonPersistentTopic = "non-persistent://prop/ns-test/topic2_" + UUID.randomUUID();
+ public void testMetricsPersistentTopicLoadFails() throws Exception {
+ final String namespace = "prop/" + UUID.randomUUID().toString().replaceAll("-", "");
+ String topic = "persistent://" + namespace + "/topic1_" + UUID.randomUUID();
+ admin.namespaces().createNamespace(namespace);
+ admin.topics().createNonPartitionedTopic(topic);
+ admin.topics().unload(topic);
+
+ // Inject an error that makes the topic load fails.
+ AtomicBoolean failMarker = new AtomicBoolean(true);
+ mockZooKeeper.failConditional(KeeperException.Code.NODEEXISTS, (op, path) -> {
+ if (failMarker.get() && op.equals(MockZooKeeper.Op.SET) &&
+ path.endsWith(TopicName.get(topic).getPersistenceNamingEncoding())) {
+ return true;
+ }
+ return false;
+ });
- BrokerService brokerService = pulsar.getBrokerService();
- brokerService = Mockito.spy(brokerService);
- // mock create persistent topic failed
- Mockito
- .doAnswer(invocation -> {
- CompletableFuture<ManagedLedgerConfig> f = new CompletableFuture<>();
- f.completeExceptionally(new RuntimeException("This is an exception"));
- return f;
- })
- .when(brokerService).getManagedLedgerConfig(Mockito.eq(TopicName.get(persistentTopic)));
-
- // mock create non-persistent topic failed
- Mockito
- .doAnswer(inv -> {
- CompletableFuture<Void> f = new CompletableFuture<>();
- f.completeExceptionally(new RuntimeException("This is an exception"));
- return f;
- })
- .when(brokerService).checkTopicNsOwnership(Mockito.eq(nonPersistentTopic));
-
-
- PulsarService pulsarService = pulsar;
- Field field = PulsarService.class.getDeclaredField("brokerService");
- field.setAccessible(true);
- field.set(pulsarService, brokerService);
-
- CompletableFuture<Producer<String>> producer = pulsarClient.newProducer(Schema.STRING)
- .topic(persistentTopic)
- .createAsync();
- CompletableFuture<Producer<String>> producer1 = pulsarClient.newProducer(Schema.STRING)
- .topic(nonPersistentTopic)
- .createAsync();
-
- producer.whenComplete((v, t) -> {
- if (t == null) {
- try {
- v.close();
- } catch (PulsarClientException e) {
- // ignore
+ // Do test
+ Thread.sleep(1000 * 3);
+ CompletableFuture<Producer<byte[]>> producer = pulsarClient.newProducer().topic(topic).createAsync();
+ JerseyClient httpClient = JerseyClientBuilder.createClient();
+ Awaitility.await().until(() -> {
+ String response = httpClient.target(pulsar.getWebServiceAddress()).path("/metrics/")
+ .request().get(String.class);
+ BufferedReader reader = new BufferedReader(new StringReader(response));
+ String line;
+ String metricsLine = null;
+ while ((line = reader.readLine()) != null) {
+ if (StringUtils.isBlank(line)) {
+ continue;
+ }
+ if (line.startsWith("#")) {
+ continue;
+ }
+ if (line.contains("topic_load_failed")) {
+ metricsLine = line;
+ break;
}
}
+ log.info("topic_load_failed: {}", metricsLine);
+ if (metricsLine == null) {
+ return false;
+ }
+ reader.close();
+ String[] parts = metricsLine.split(" ");
+ Double value = Double.valueOf(parts[parts.length - 1]);
+ return value >= 1D;
});
- producer1.whenComplete((v, t) -> {
- if (t == null) {
- try {
- v.close();
- } catch (PulsarClientException e) {
- // ignore
+
+ // Remove the injection.
+ failMarker.set(false);
+ // cleanup.
+ httpClient.close();
+ producer.join().close();
+ admin.topics().delete(topic);
+ admin.namespaces().deleteNamespace(namespace);
+ }
+
+ @Test
+ public void testMetricsNonPersistentTopicLoadFails() throws Exception {
+ final String namespace = "prop/" + UUID.randomUUID().toString().replaceAll("-", "");
+ String topic = "non-persistent://" + namespace + "/topic1_" + UUID.randomUUID();
+ admin.namespaces().createNamespace(namespace);
+
+ // Inject an error that makes the topic load fails.
+ // Since we did not set a topic factory name, the "topicFactory" variable is null, inject a mocked
+ // "topicFactory".
+ Field fieldTopicFactory = BrokerService.class.getDeclaredField("topicFactory");
+ fieldTopicFactory.setAccessible(true);
+ TopicFactory originalTopicFactory = (TopicFactory) fieldTopicFactory.get(pulsar.getBrokerService());
+ assertNull(originalTopicFactory);
+ TopicFactory mockedTopicFactory = mock(TopicFactory.class);
+ when(mockedTopicFactory.create(anyString(), any(), any(), any()))
+ .thenThrow(new RuntimeException("mocked error"));
+ fieldTopicFactory.set(pulsar.getBrokerService(), mockedTopicFactory);
+
+ // Do test.
+ CompletableFuture<Producer<byte[]>> producer = pulsarClient.newProducer().topic(topic).createAsync();
+ JerseyClient httpClient = JerseyClientBuilder.createClient();
+ Awaitility.await().until(() -> {
+ String response = httpClient.target(pulsar.getWebServiceAddress()).path("/metrics/")
+ .request().get(String.class);
+ BufferedReader reader = new BufferedReader(new StringReader(response));
+ String line;
+ String metricsLine = null;
+ while ((line = reader.readLine()) != null) {
+ if (StringUtils.isBlank(line)) {
+ continue;
+ }
+ if (line.startsWith("#")) {
+ continue;
}
+ if (line.contains("topic_load_failed")) {
+ metricsLine = line;
+ break;
+ }
+ }
+ log.info("topic_load_failed: {}", metricsLine);
+ if (metricsLine == null) {
+ return false;
}
+ reader.close();
+ String[] parts = metricsLine.split(" ");
+ Double value = Double.valueOf(parts[parts.length - 1]);
+ return value >= 1D;
});
- Awaitility.waitAtMost(2, TimeUnit.MINUTES).until(() -> {
- String json = admin.brokerStats().getMetrics();
- JsonArray metrics = new Gson().fromJson(json, JsonArray.class);
- AtomicBoolean flag = new AtomicBoolean(false);
-
- metrics.forEach(ele -> {
- JsonObject obj = ((JsonObject) ele);
- JsonObject metrics0 = (JsonObject) obj.get("metrics");
- JsonPrimitive v = (JsonPrimitive) metrics0.get("brk_topic_load_failed_count");
- if (null != v && v.getAsDouble() >= 2D) {
- flag.set(true);
- }
- });
+ // Remove the injection.
+ fieldTopicFactory.set(pulsar.getBrokerService(), null);
- return flag.get();
- });
+ // cleanup.
+ httpClient.close();
+ producer.join().close();
+ admin.topics().delete(topic);
+ admin.namespaces().deleteNamespace(namespace);
}
}
(pulsar) 03/03: [cleanup] [test] Cleanup unnecessary imports
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a8cd7befe6ce7258b16e94764cb347ff5c6d600f
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu May 2 23:26:33 2024 +0800
[cleanup] [test] Cleanup unnecessary imports
---
.../test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java | 4 ----
1 file changed, 4 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 70919716df7..121f5c7bf9a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -33,7 +33,6 @@ import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
-import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
@@ -45,12 +44,10 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
-import java.io.Reader;
import java.io.StringReader;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -87,7 +84,6 @@ import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient;
import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
import org.apache.pulsar.client.admin.BrokerStats;
import org.apache.pulsar.client.admin.PulsarAdminException;
(pulsar) 01/03: [improve][broker] Add `topic_load_failed` metric (#19236)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a03543a4e4867835152ccae9a0ce5bb7b9042c42
Author: Tao Jiuming <95...@users.noreply.github.com>
AuthorDate: Sat Jun 3 21:14:16 2023 +0800
[improve][broker] Add `topic_load_failed` metric (#19236)
Co-authored-by: daojun <jm...@streamnative.io>
(cherry picked from commit 43b3622cc7e7f746ca9920fdc704dc7448767ac7)
---
.../pulsar/broker/service/BrokerService.java | 9 +++
.../apache/pulsar/broker/service/PulsarStats.java | 4 ++
.../broker/stats/BrokerOperabilityMetrics.java | 10 ++-
.../pulsar/broker/service/BrokerServiceTest.java | 79 ++++++++++++++++++++++
.../pulsar/broker/stats/PrometheusMetricsTest.java | 4 ++
5 files changed, 105 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b921c4b5dc6..65366f7eb22 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1183,6 +1183,10 @@ public class BrokerService implements Closeable {
private CompletableFuture<Optional<Topic>> createNonPersistentTopic(String topic) {
CompletableFuture<Optional<Topic>> topicFuture = new CompletableFuture<>();
+ topicFuture.exceptionally(t -> {
+ pulsarStats.recordTopicLoadFailed();
+ return null;
+ });
if (!pulsar.getConfiguration().isEnableNonPersistentTopics()) {
if (log.isDebugEnabled()) {
log.debug("Broker is unable to load non-persistent topic {}", topic);
@@ -1538,6 +1542,11 @@ public class BrokerService implements Closeable {
TopicName topicName = TopicName.get(topic);
final long topicCreateTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
+ topicFuture.exceptionally(t -> {
+ pulsarStats.recordTopicLoadFailed();
+ return null;
+ });
+
if (isTransactionInternalName(topicName)) {
String msg = String.format("Can not create transaction system topic %s", topic);
log.warn(msg);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java
index ff74cf839ae..69cfc68129c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java
@@ -239,6 +239,10 @@ public class PulsarStats implements Closeable {
}
}
+ public void recordTopicLoadFailed() {
+ brokerOperabilityMetrics.recordTopicLoadFailed();
+ }
+
public void recordConnectionCreate() {
brokerOperabilityMetrics.recordConnectionCreate();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
index 4fd9e35dd7b..c0810f47278 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.stats;
import com.google.common.collect.Maps;
+import io.prometheus.client.Counter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -29,6 +30,7 @@ import org.apache.pulsar.common.stats.Metrics;
/**
*/
public class BrokerOperabilityMetrics {
+ private static final Counter TOPIC_LOAD_FAILED = Counter.build("topic_load_failed", "-").register();
private final List<Metrics> metricsList;
private final String localCluster;
private final DimensionStats topicLoadStats;
@@ -90,7 +92,9 @@ public class BrokerOperabilityMetrics {
}
Metrics getTopicLoadMetrics() {
- return getDimensionMetrics("topic_load_times", "topic_load", topicLoadStats);
+ Metrics metrics = getDimensionMetrics("topic_load_times", "topic_load", topicLoadStats);
+ metrics.put("brk_topic_load_failed_count", TOPIC_LOAD_FAILED.get());
+ return metrics;
}
Metrics getZkWriteLatencyMetrics() {
@@ -136,6 +140,10 @@ public class BrokerOperabilityMetrics {
zkReadLatencyStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS);
}
+ public void recordTopicLoadFailed() {
+ this.TOPIC_LOAD_FAILED.inc();
+ }
+
public void recordConnectionCreate() {
this.connectionTotalCreatedCount.increment();
this.connectionActive.increment();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 5a54473247f..ea10b24b385 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -34,6 +34,7 @@ import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
@@ -1628,4 +1629,82 @@ public class BrokerServiceTest extends BrokerTestBase {
assertTrue(conf.isForceDeleteTenantAllowed());
});
}
+
+
+ @Test
+ public void testBrokerStatsTopicLoadFailed() throws Exception {
+ admin.namespaces().createNamespace("prop/ns-test");
+
+ String persistentTopic = "persistent://prop/ns-test/topic1_" + UUID.randomUUID();
+ String nonPersistentTopic = "non-persistent://prop/ns-test/topic2_" + UUID.randomUUID();
+
+ BrokerService brokerService = pulsar.getBrokerService();
+ brokerService = Mockito.spy(brokerService);
+ // mock create persistent topic failed
+ Mockito
+ .doAnswer(invocation -> {
+ CompletableFuture<ManagedLedgerConfig> f = new CompletableFuture<>();
+ f.completeExceptionally(new RuntimeException("This is an exception"));
+ return f;
+ })
+ .when(brokerService).getManagedLedgerConfig(Mockito.eq(TopicName.get(persistentTopic)));
+
+ // mock create non-persistent topic failed
+ Mockito
+ .doAnswer(inv -> {
+ CompletableFuture<Void> f = new CompletableFuture<>();
+ f.completeExceptionally(new RuntimeException("This is an exception"));
+ return f;
+ })
+ .when(brokerService).checkTopicNsOwnership(Mockito.eq(nonPersistentTopic));
+
+
+ PulsarService pulsarService = pulsar;
+ Field field = PulsarService.class.getDeclaredField("brokerService");
+ field.setAccessible(true);
+ field.set(pulsarService, brokerService);
+
+ CompletableFuture<Producer<String>> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(persistentTopic)
+ .createAsync();
+ CompletableFuture<Producer<String>> producer1 = pulsarClient.newProducer(Schema.STRING)
+ .topic(nonPersistentTopic)
+ .createAsync();
+
+ producer.whenComplete((v, t) -> {
+ if (t == null) {
+ try {
+ v.close();
+ } catch (PulsarClientException e) {
+ // ignore
+ }
+ }
+ });
+ producer1.whenComplete((v, t) -> {
+ if (t == null) {
+ try {
+ v.close();
+ } catch (PulsarClientException e) {
+ // ignore
+ }
+ }
+ });
+
+ Awaitility.waitAtMost(2, TimeUnit.MINUTES).until(() -> {
+ String json = admin.brokerStats().getMetrics();
+ JsonArray metrics = new Gson().fromJson(json, JsonArray.class);
+ AtomicBoolean flag = new AtomicBoolean(false);
+
+ metrics.forEach(ele -> {
+ JsonObject obj = ((JsonObject) ele);
+ JsonObject metrics0 = (JsonObject) obj.get("metrics");
+ JsonPrimitive v = (JsonPrimitive) metrics0.get("brk_topic_load_failed_count");
+ if (null != v && v.getAsDouble() >= 2D) {
+ flag.set(true);
+ }
+ });
+
+ return flag.get();
+ });
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index 1fa523b7be9..8a07038c043 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -323,6 +323,10 @@ public class PrometheusMetricsTest extends BrokerTestBase {
assertEquals(cm.size(), 1);
assertEquals(cm.get(0).tags.get("cluster"), "test");
+ cm = (List<Metric>) metrics.get("topic_load_failed_total");
+ assertEquals(cm.size(), 1);
+ assertEquals(cm.get(0).tags.get("cluster"), "test");
+
cm = (List<Metric>) metrics.get("pulsar_in_bytes_total");
assertEquals(cm.size(), 2);
assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");