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 2026/10/16 20:22:26 UTC
(pulsar) 01/30: [improve][broker] Add reason label for topic_load_failed_total (#26264)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2-backports-2026-10-16
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c2c078da2301fb43745f818830ab09cc60a4372c
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Aug 24 04:46:51 2026 +0800
[improve][broker] Add reason label for topic_load_failed_total (#26264)
(cherry picked from commit 53b6f623bd0ca809184609f5bb00c37186723056)
---
pip/pip-382.md | 40 +++++++-
.../pulsar/broker/service/BrokerService.java | 108 +++++++++++++++++----
.../apache/pulsar/broker/service/PulsarStats.java | 5 +-
.../pulsar/broker/service/TopicLoadingContext.java | 65 +++++++++++++
.../broker/service/persistent/PersistentTopic.java | 21 +++-
.../broker/stats/BrokerOperabilityMetrics.java | 45 ++++++++-
.../pulsar/broker/service/BrokerServiceTest.java | 83 +++++++++++++++-
.../pulsar/broker/stats/PrometheusMetricsTest.java | 7 +-
8 files changed, 340 insertions(+), 34 deletions(-)
diff --git a/pip/pip-382.md b/pip/pip-382.md
index adc6e636fe49..4d610e65cb3c 100644
--- a/pip/pip-382.md
+++ b/pip/pip-382.md
@@ -4,7 +4,7 @@
Pulsar has a metric that indicates load topic failed: `topic_load_failed_total`, it will be increased at the following cases
- The target bundle in unloading.
-- Failed to load policies.
+- Failed to load namespace-level or topic-level policies.
- Failed to load up Managed Ledger.
- Failed to read Metadata store.
- Topic initialize fails, such as failed to re-build deduplication info.
@@ -21,13 +21,47 @@ Add a label named reason for topic_load_failed_total
- label name: `reason`
- label values:
- `bundle_unloading`
- - `failed_load_policies`
+ - `failed_load_namespace_policies`
+ - `failed_load_topic_policies`
- `failed_load_ml`
+ - `failed_check_ownership`
- `failed_access_metadata_store`
- `failed_init`
- - `timeout`
+ - `timeout` (generic timeout, or timeout while more than one loading stage is pending)
+ - `timeout_load_namespace_policies`
+ - `timeout_load_topic_policies`
+ - `timeout_load_ml`
+ - `timeout_init`
+ - `timeout_dedup`
- `others`
+### Implementation Notes
+
+Each persistent topic-loading operation records at most one failure. The failure reason is attached at the
+operation that can identify it, rather than by wrapping exceptions solely for metrics. The final topic future
+increments the counter once, so callers that reuse an in-flight load future do not create additional samples.
+
+Timeouts describe the loading stage that is still pending when the timeout fires:
+
+| Pending stage | Timeout reason |
+|---|---|
+| namespace-level policy loading, including during `PersistentTopic` initialization | `timeout_load_namespace_policies` |
+| topic-level policy loading, including during `PersistentTopic` initialization | `timeout_load_topic_policies` |
+| opening the managed ledger | `timeout_load_ml` |
+| topic initialization, replication, or compacted-subscription setup | `timeout_init` |
+| deduplication initialization | `timeout_dedup` |
+| no single pending stage | `timeout` |
+
+`bundle_unloading` includes failures while a bundle is unloading and failures because this broker is not the
+topic owner. `TopicMigratedException` is an expected migration control flow and does not increment
+`topic_load_failed_total`.
+
+`failed_check_ownership` covers failures while resolving the namespace bundle or checking its ownership. A
+successful ownership check that determines this broker is not the owner is reported as `bundle_unloading`.
+
+`brk_topic_load_failed_count` remains available for compatibility and is the sum of all reason-labelled failure
+series.
+
# Monitoring & Alternatives
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 5c3974253020..660238b154c0 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
@@ -128,6 +128,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotRea
import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedException;
import org.apache.pulsar.broker.service.TopicEventsListener.EventStage;
import org.apache.pulsar.broker.service.TopicEventsListener.TopicEvent;
+import org.apache.pulsar.broker.service.TopicLoadingContext.TopicLoadingStage;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentSystemTopic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
@@ -137,6 +138,7 @@ import org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryCla
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.SystemTopic;
import org.apache.pulsar.broker.service.plugin.EntryFilterProvider;
+import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics.TopicLoadFailureReason;
import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
import org.apache.pulsar.broker.stats.prometheus.metrics.ObserverGauge;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
@@ -1248,12 +1250,13 @@ public class BrokerService implements Closeable {
}
topicFuture.exceptionally(t -> {
final var latency = context.traceAndGetLatency("fail").description();
- if (FutureUtil.unwrapCompletionException(t) instanceof TimeoutException) {
+ final var unwrapped = FutureUtil.unwrapCompletionException(t);
+ if (unwrapped instanceof TimeoutException) {
log.warn("Failed to load {} within {} s (latency: {})", topicName, timeoutSeconds, latency);
} else {
log.warn("Failed to load {} (latency: {})", topicName, latency, t);
}
- pulsarStats.recordTopicLoadFailed();
+ recordTopicLoadFailure(context, unwrapped);
return Optional.empty();
});
context.trace("topic exists", checkNonPartitionedTopicExists(topicName)).thenAccept(exists -> {
@@ -1263,7 +1266,7 @@ public class BrokerService implements Closeable {
}
// The topic level policies are not needed now, but the meaning of calling
// "getTopicPoliciesBypassSystemTopic" will wait for system topic policies initialization.
- final var systemTopicLoadFuture = context.trace("system topic",
+ final var systemTopicLoadFuture = context.trace(TopicLoadingStage.TOPIC_POLICIES,
getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY));
systemTopicLoadFuture.thenRun(() -> {
final var inserted = new MutableBoolean(false);
@@ -1288,10 +1291,12 @@ public class BrokerService implements Closeable {
}).exceptionally(e -> {
log.warn("[{}] Topic creation encountered an exception"
+ " by initialize topic policies service", topicName);
+ context.setTopicLoadFailureReason(TopicLoadFailureReason.FAILED_LOAD_TOPIC_POLICIES);
failTopicFuture(topicName.toString(), topicFuture, e);
return null;
});
}).exceptionally(e -> {
+ context.setTopicLoadFailureReason(TopicLoadFailureReason.FAILED_ACCESS_METADATA_STORE);
failTopicFuture(topicName.toString(), topicFuture, e);
return null;
});
@@ -1351,6 +1356,20 @@ public class BrokerService implements Closeable {
topicFuture.completeExceptionally(rc);
}
+ private void recordTopicLoadFailure(TopicLoadingContext context, Throwable throwable) {
+ if (throwable instanceof TopicMigratedException) {
+ return;
+ }
+ if (throwable instanceof TimeoutException) {
+ pulsarStats.recordTopicLoadFailed(context.getTopicLoadTimeoutReason());
+ } else if (throwable instanceof ServiceUnitNotReadyException) {
+ pulsarStats.recordTopicLoadFailed(TopicLoadFailureReason.BUNDLE_UNLOADING);
+ } else {
+ TopicLoadFailureReason reason = context.getTopicLoadFailureReason();
+ pulsarStats.recordTopicLoadFailed(reason != null ? reason : TopicLoadFailureReason.OTHERS);
+ }
+ }
+
private CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesBypassSystemTopic(@NonNull TopicName topicName,
TopicPoliciesService.GetType type) {
if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
@@ -1359,6 +1378,33 @@ public class BrokerService implements Closeable {
return pulsar.getTopicPoliciesService().getTopicPoliciesAsync(topicName, type);
}
+ private <T> CompletableFuture<T> trackTopicLoadFailure(TopicLoadingContext context,
+ TopicLoadingStage stage,
+ TopicLoadFailureReason reason,
+ CompletableFuture<T> future) {
+ if (context == null) {
+ return future;
+ }
+ return context.trace(stage, future).whenComplete((__, throwable) -> {
+ if (throwable != null) {
+ context.setTopicLoadFailureReason(reason);
+ }
+ });
+ }
+
+ private <T> CompletableFuture<T> trackTopicLoadFailure(TopicLoadingContext context,
+ TopicLoadFailureReason reason,
+ CompletableFuture<T> future) {
+ if (context == null) {
+ return future;
+ }
+ return future.whenComplete((__, throwable) -> {
+ if (throwable != null) {
+ context.setTopicLoadFailureReason(reason);
+ }
+ });
+ }
+
public CompletableFuture<Void> deleteTopic(String topic, boolean forceDelete) {
topicEventsDispatcher.notify(topic, TopicEvent.DELETE, EventStage.BEFORE);
CompletableFuture<Void> result = deleteTopicInternal(topic, forceDelete);
@@ -1507,7 +1553,7 @@ public class BrokerService implements Closeable {
private CompletableFuture<Optional<Topic>> createNonPersistentTopic(String topic) {
CompletableFuture<Optional<Topic>> topicFuture = new CompletableFuture<>();
topicFuture.exceptionally(t -> {
- pulsarStats.recordTopicLoadFailed();
+ pulsarStats.recordTopicLoadFailed(TopicLoadFailureReason.OTHERS);
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
return null;
});
@@ -1789,7 +1835,9 @@ public class BrokerService implements Closeable {
protected CompletableFuture<Optional<Topic>> loadOrCreatePersistentTopic(TopicLoadingContext context) {
final var topic = context.getTopicName().toString();
final var topicFuture = context.getTopicFuture();
- context.trace("ownership", checkTopicNsOwnership(topic))
+ // ServiceUnitNotReadyException is classified as bundle_unloading when the topic future completes.
+ trackTopicLoadFailure(context, TopicLoadFailureReason.FAILED_CHECK_OWNERSHIP,
+ context.trace("ownership", checkTopicNsOwnership(topic)))
.thenRun(() -> {
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
@@ -1874,7 +1922,9 @@ public class BrokerService implements Closeable {
TopicName topicName = context.getTopicName();
final var topic = topicName.toString();
final var topicFuture = context.getTopicFuture();
- context.trace("2nd ownership", checkTopicNsOwnership(topic)).thenRun(() -> {
+ // ServiceUnitNotReadyException is classified as bundle_unloading when the topic future completes.
+ trackTopicLoadFailure(context, TopicLoadFailureReason.FAILED_CHECK_OWNERSHIP,
+ context.trace("2nd ownership", checkTopicNsOwnership(topic))).thenRun(() -> {
CompletableFuture<Map<String, String>> propertiesFuture;
if (context.getProperties() == null) {
//Read properties from storage when loading topic.
@@ -1887,6 +1937,7 @@ public class BrokerService implements Closeable {
//TODO add topicName in properties?
createPersistentTopic0(context);
}).exceptionally(throwable -> {
+ context.setTopicLoadFailureReason(TopicLoadFailureReason.FAILED_ACCESS_METADATA_STORE);
failTopicFuture(topic, topicFuture, throwable);
return null;
});
@@ -1916,8 +1967,9 @@ public class BrokerService implements Closeable {
CompletableFuture<Void> isTopicAlreadyMigrated = checkTopicAlreadyMigrated(topicName);
maxTopicsCheck.thenCompose(partitionedTopicMetadata -> validateTopicConsistency(topicName))
.thenCompose(__ -> isTopicAlreadyMigrated)
- .thenCompose(__ -> getManagedLedgerConfig(topicName))
- .thenCombine(pulsar().getNamespaceService().checkTopicExistsAsync(topicName).thenApply(n -> {
+ .thenCompose(__ -> getManagedLedgerConfig(topicName, context))
+ .thenCombine(trackTopicLoadFailure(context, TopicLoadFailureReason.FAILED_ACCESS_METADATA_STORE,
+ pulsar().getNamespaceService().checkTopicExistsAsync(topicName)).thenApply(n -> {
boolean found = n.isExists();
n.recycle();
return found;
@@ -1989,22 +2041,23 @@ public class BrokerService implements Closeable {
// Once we have the configuration, we can proceed with the async open operation
ManagedLedgerFactory managedLedgerFactory =
getManagedLedgerFactoryForTopic(topicName, managedLedgerConfig.getStorageClassName());
+ context.start(TopicLoadingStage.OPEN_ML);
managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(), managedLedgerConfig,
new OpenLedgerCallback() {
@Override
public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
try {
- context.trace("open-ml");
+ context.finish(TopicLoadingStage.OPEN_ML);
PersistentTopic persistentTopic = isSystemTopic(topic)
? new SystemTopic(topic, ledger, BrokerService.this)
: newTopic(topic, ledger, BrokerService.this, PersistentTopic.class);
persistentTopic.setCreateFuture(topicFuture);
- context.trace("init", persistentTopic.initialize())
- .thenCompose(__ -> context.trace("pre-create compacted sub",
+ context.trace(TopicLoadingStage.INITIALIZE, persistentTopic.initialize(context))
+ .thenCompose(__ -> context.trace(TopicLoadingStage.PRE_CREATE_COMPACTED_SUB,
persistentTopic.preCreateSubscriptionForCompactionIfNeeded()))
- .thenCompose(__ -> context.trace("replication",
+ .thenCompose(__ -> context.trace(TopicLoadingStage.REPLICATION,
persistentTopic.initializeCheckReplication()))
- .thenCompose(v -> context.trace("deduplication",
+ .thenCompose(v -> context.trace(TopicLoadingStage.DEDUPLICATION,
persistentTopic.checkDeduplicationStatus()))
.thenRun(() -> {
final var latency = context.traceAndGetLatency("done");
@@ -2045,6 +2098,8 @@ public class BrokerService implements Closeable {
log.warn("[{}] Get an error when closing topic.",
topic, closeEx);
}
+ context.setTopicLoadFailureReason(
+ TopicLoadFailureReason.FAILED_INIT);
topicFuture.completeExceptionally(ex);
});
});
@@ -2057,12 +2112,14 @@ public class BrokerService implements Closeable {
@Override
public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ context.finish(TopicLoadingStage.OPEN_ML);
if (!createIfMissing && exception instanceof ManagedLedgerNotFoundException) {
// We were just trying to load a topic and the topic doesn't exist
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
loadFuture.completeExceptionally(exception);
topicFuture.complete(Optional.empty());
} else {
+ context.setTopicLoadFailureReason(TopicLoadFailureReason.FAILED_LOAD_ML);
failTopicFuture(topic, topicFuture, new PersistenceException(exception));
}
}
@@ -2098,6 +2155,11 @@ public class BrokerService implements Closeable {
}
public CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@NonNull TopicName topicName) {
+ return getManagedLedgerConfig(topicName, null);
+ }
+
+ private CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(
+ @NonNull TopicName topicName, @Nullable TopicLoadingContext context) {
if (topicName == null) {
return FutureUtil.failedFuture(new NullPointerException("topicName"));
}
@@ -2107,11 +2169,19 @@ public class BrokerService implements Closeable {
NamespaceResources nsr = pulsar.getPulsarResources().getNamespaceResources();
LocalPoliciesResources lpr = pulsar.getPulsarResources().getLocalPolicies();
final CompletableFuture<Optional<TopicPolicies>> topicPoliciesFuture =
- getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY);
+ trackTopicLoadFailure(context, TopicLoadingStage.TOPIC_POLICIES,
+ TopicLoadFailureReason.FAILED_LOAD_TOPIC_POLICIES,
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY));
final CompletableFuture<Optional<TopicPolicies>> globalTopicPoliciesFuture =
- getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.GLOBAL_ONLY);
- final CompletableFuture<Optional<Policies>> nsPolicies = nsr.getPoliciesAsync(namespace);
- final CompletableFuture<Optional<LocalPolicies>> lcPolicies = lpr.getLocalPoliciesAsync(namespace);
+ trackTopicLoadFailure(context, TopicLoadingStage.TOPIC_POLICIES,
+ TopicLoadFailureReason.FAILED_LOAD_TOPIC_POLICIES,
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.GLOBAL_ONLY));
+ final CompletableFuture<Optional<Policies>> nsPolicies = trackTopicLoadFailure(context,
+ TopicLoadingStage.NAMESPACE_POLICIES, TopicLoadFailureReason.FAILED_LOAD_NAMESPACE_POLICIES,
+ nsr.getPoliciesAsync(namespace));
+ final CompletableFuture<Optional<LocalPolicies>> lcPolicies = trackTopicLoadFailure(context,
+ TopicLoadingStage.NAMESPACE_POLICIES, TopicLoadFailureReason.FAILED_LOAD_NAMESPACE_POLICIES,
+ lpr.getLocalPoliciesAsync(namespace));
return topicPoliciesFuture.thenCombine(globalTopicPoliciesFuture, (topicP, globalTopicP) -> {
return new ImmutablePair<>(topicP, globalTopicP);
}).thenCombine(nsPolicies, (topicPoliciesPair, np) -> {
@@ -3564,7 +3634,9 @@ public class BrokerService implements Closeable {
pendingTopic.trace("queued");
final String topic = pendingTopic.getTopicName().toString();
- pendingTopic.trace("ownership", checkTopicNsOwnership(topic)).thenRun(() -> {
+ // ServiceUnitNotReadyException is classified as bundle_unloading when the topic future completes.
+ trackTopicLoadFailure(pendingTopic, TopicLoadFailureReason.FAILED_CHECK_OWNERSHIP,
+ pendingTopic.trace("ownership", checkTopicNsOwnership(topic))).thenRun(() -> {
CompletableFuture<Optional<Topic>> pendingFuture = pendingTopic.getTopicFuture();
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
final boolean acquiredPermit = topicLoadSemaphore.tryAcquire();
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 45a0e8b42f0c..6a7c8455c279 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
@@ -34,6 +34,7 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics;
+import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics.TopicLoadFailureReason;
import org.apache.pulsar.broker.stats.BrokerStats;
import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
import org.apache.pulsar.broker.stats.NamespaceStats;
@@ -262,8 +263,8 @@ public class PulsarStats implements Closeable {
}
}
- public void recordTopicLoadFailed() {
- brokerOperabilityMetrics.recordTopicLoadFailed();
+ public void recordTopicLoadFailed(TopicLoadFailureReason reason) {
+ brokerOperabilityMetrics.recordTopicLoadFailed(reason);
}
public void recordConnectionCreate() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
index b9a668f92b26..a9bad04b1820 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
@@ -21,15 +21,37 @@ package org.apache.pulsar.broker.service;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Getter;
import lombok.Setter;
+import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics.TopicLoadFailureReason;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.LatencyTracer;
import org.jspecify.annotations.Nullable;
public class TopicLoadingContext extends LatencyTracer {
+ public enum TopicLoadingStage {
+ NAMESPACE_POLICIES("namespace policies", TopicLoadFailureReason.TIMEOUT_LOAD_NAMESPACE_POLICIES),
+ TOPIC_POLICIES("topic policies", TopicLoadFailureReason.TIMEOUT_LOAD_TOPIC_POLICIES),
+ OPEN_ML("open-ml", TopicLoadFailureReason.TIMEOUT_LOAD_ML),
+ INITIALIZE("init", TopicLoadFailureReason.TIMEOUT_INIT),
+ PRE_CREATE_COMPACTED_SUB("pre-create compacted sub", TopicLoadFailureReason.TIMEOUT_INIT),
+ REPLICATION("replication", TopicLoadFailureReason.TIMEOUT_INIT),
+ DEDUPLICATION("deduplication", TopicLoadFailureReason.TIMEOUT_DEDUP);
+
+ private final String tracePoint;
+ private final TopicLoadFailureReason timeoutReason;
+
+ TopicLoadingStage(String tracePoint, TopicLoadFailureReason timeoutReason) {
+ this.tracePoint = tracePoint;
+ this.timeoutReason = timeoutReason;
+ }
+ }
+
@Getter
private final TopicName topicName;
@Getter
@@ -39,6 +61,8 @@ public class TopicLoadingContext extends LatencyTracer {
@Getter
@Setter
@Nullable private Map<String, String> properties;
+ private final AtomicReference<TopicLoadFailureReason> failureReason = new AtomicReference<>();
+ private final ConcurrentHashMap<TopicLoadingStage, AtomicInteger> pendingStages = new ConcurrentHashMap<>();
public TopicLoadingContext(TopicName topicName, boolean createIfMissing,
CompletableFuture<Optional<Topic>> topicFuture) {
@@ -48,4 +72,45 @@ public class TopicLoadingContext extends LatencyTracer {
this.createIfMissing = createIfMissing;
this.topicFuture = topicFuture;
}
+
+ public <T> CompletableFuture<T> trace(TopicLoadingStage stage, CompletableFuture<T> future) {
+ if (future.isDone()) {
+ return future;
+ }
+ start(stage);
+ return future.whenComplete((__, ___) -> {
+ finish(stage);
+ });
+ }
+
+ public void start(TopicLoadingStage stage) {
+ pendingStages.computeIfAbsent(stage, __ -> new AtomicInteger()).incrementAndGet();
+ }
+
+ public void finish(TopicLoadingStage stage) {
+ pendingStages.computeIfPresent(stage, (__, count) -> count.decrementAndGet() == 0 ? null : count);
+ trace(stage.tracePoint);
+ }
+
+ public void setTopicLoadFailureReason(TopicLoadFailureReason reason) {
+ failureReason.compareAndSet(null, reason);
+ }
+
+ public TopicLoadFailureReason getTopicLoadFailureReason() {
+ return failureReason.get();
+ }
+
+ public TopicLoadFailureReason getTopicLoadTimeoutReason() {
+ boolean namespacePoliciesPending = pendingStages.containsKey(TopicLoadingStage.NAMESPACE_POLICIES);
+ boolean topicPoliciesPending = pendingStages.containsKey(TopicLoadingStage.TOPIC_POLICIES);
+ if (namespacePoliciesPending != topicPoliciesPending) {
+ return namespacePoliciesPending
+ ? TopicLoadFailureReason.TIMEOUT_LOAD_NAMESPACE_POLICIES
+ : TopicLoadFailureReason.TIMEOUT_LOAD_TOPIC_POLICIES;
+ }
+ if (pendingStages.size() != 1) {
+ return TopicLoadFailureReason.TIMEOUT;
+ }
+ return pendingStages.keySet().iterator().next().timeoutReason;
+ }
}
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 aaedea324789..78ee5d27fa35 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
@@ -133,6 +133,8 @@ import org.apache.pulsar.broker.service.StreamingStats;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.SubscriptionOption;
import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.TopicLoadingContext;
+import org.apache.pulsar.broker.service.TopicLoadingContext.TopicLoadingStage;
import org.apache.pulsar.broker.service.TopicPoliciesService;
import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage;
@@ -206,6 +208,7 @@ import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.utils.StatsOutputStream;
import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -482,16 +485,25 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
@Override
public CompletableFuture<Void> initialize() {
+ return initialize(null);
+ }
+
+ public CompletableFuture<Void> initialize(@Nullable TopicLoadingContext loadingContext) {
List<CompletableFuture<Void>> futures = new ArrayList<>();
futures.add(brokerService.getPulsar().newTopicCompactionService(topic).thenAccept(service -> {
PersistentTopic.this.topicCompactionService = service;
this.createPersistentSubscriptions();
}));
+ CompletableFuture<Optional<Policies>> namespacePoliciesFuture = brokerService.pulsar().getPulsarResources()
+ .getNamespaceResources().getPoliciesAsync(TopicName.get(topic).getNamespaceObject());
+ if (loadingContext != null) {
+ namespacePoliciesFuture = loadingContext.trace(TopicLoadingStage.NAMESPACE_POLICIES,
+ namespacePoliciesFuture);
+ }
+ final CompletableFuture<Optional<Policies>> trackedNamespacePoliciesFuture = namespacePoliciesFuture;
return FutureUtil.waitForAll(futures).thenCompose(__ ->
- brokerService.pulsar().getPulsarResources().getNamespaceResources()
- .getPoliciesAsync(TopicName.get(topic).getNamespaceObject())
- .thenAcceptAsync(optPolicies -> {
+ trackedNamespacePoliciesFuture.thenAcceptAsync(optPolicies -> {
if (!optPolicies.isPresent()) {
isEncryptionRequired = false;
updatePublishRateLimiter();
@@ -517,7 +529,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
isAllowAutoUpdateSchema = policies.is_allow_auto_update_schema;
}, getPoliciesNotifyThread())
- .thenCompose(ignore -> initTopicPolicy())
+ .thenCompose(ignore -> loadingContext == null ? initTopicPolicy()
+ : loadingContext.trace(TopicLoadingStage.TOPIC_POLICIES, initTopicPolicy()))
.thenCompose(ignore -> removeOrphanReplicationCursors()));
}
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 33aca625e66b..2c6d3bd60cdf 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
@@ -38,13 +38,48 @@ import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric;
/**
*/
public class BrokerOperabilityMetrics implements AutoCloseable {
- private static final Counter TOPIC_LOAD_FAILED = Counter.build("topic_load_failed", "-").register();
+ public enum TopicLoadFailureReason {
+ BUNDLE_UNLOADING("bundle_unloading"),
+ FAILED_LOAD_NAMESPACE_POLICIES("failed_load_namespace_policies"),
+ FAILED_LOAD_TOPIC_POLICIES("failed_load_topic_policies"),
+ FAILED_LOAD_ML("failed_load_ml"),
+ FAILED_CHECK_OWNERSHIP("failed_check_ownership"),
+ FAILED_ACCESS_METADATA_STORE("failed_access_metadata_store"),
+ FAILED_INIT("failed_init"),
+ TIMEOUT("timeout"),
+ TIMEOUT_LOAD_NAMESPACE_POLICIES("timeout_load_namespace_policies"),
+ TIMEOUT_LOAD_TOPIC_POLICIES("timeout_load_topic_policies"),
+ TIMEOUT_LOAD_ML("timeout_load_ml"),
+ TIMEOUT_INIT("timeout_init"),
+ TIMEOUT_DEDUP("timeout_dedup"),
+ OTHERS("others");
+
+ private final String label;
+
+ TopicLoadFailureReason(String label) {
+ this.label = label;
+ }
+
+ public String label() {
+ return label;
+ }
+ }
+
+ private static final Counter TOPIC_LOAD_FAILED = Counter.build("topic_load_failed", "-")
+ .labelNames("reason").create().register();
private static final Counter NON_RECOVERABLE_LEDGERS_SKIPPED = Counter.build(
"pulsar_broker_non_recoverable_ledgers_skipped_total",
"Count of non-recoverable ledgers skipped").register();
private static final Counter NON_RECOVERABLE_ENTRIES_SKIPPED = Counter.build(
"pulsar_broker_non_recoverable_entries_skipped_total",
"Count of non-recoverable entries skipped").register();
+
+ static {
+ for (TopicLoadFailureReason reason : TopicLoadFailureReason.values()) {
+ TOPIC_LOAD_FAILED.labels(reason.label());
+ }
+ }
+
private final List<Metrics> metricsList;
private final String localCluster;
private final DimensionStats topicLoadStats;
@@ -206,7 +241,9 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
Metrics getTopicLoadMetrics() {
Metrics metrics = getDimensionMetrics("pulsar_topic_load_times", "topic_load", topicLoadStats);
- metrics.put("brk_topic_load_failed_count", TOPIC_LOAD_FAILED.get());
+ metrics.put("brk_topic_load_failed_count", Arrays.stream(TopicLoadFailureReason.values())
+ .mapToDouble(reason -> TOPIC_LOAD_FAILED.labels(reason.label()).get())
+ .sum());
return metrics;
}
@@ -235,8 +272,8 @@ public class BrokerOperabilityMetrics implements AutoCloseable {
topicLoadStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS);
}
- public void recordTopicLoadFailed() {
- this.TOPIC_LOAD_FAILED.inc();
+ public void recordTopicLoadFailed(TopicLoadFailureReason reason) {
+ TOPIC_LOAD_FAILED.labels(reason.label()).inc();
}
public void recordConnectionCreate() {
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 b32e76a7d24f..edbeeba79151 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
@@ -88,8 +88,10 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.PulsarMetadataEventSynchronizer.State;
+import org.apache.pulsar.broker.service.TopicLoadingContext.TopicLoadingStage;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics.TopicLoadFailureReason;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsServlet;
import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
@@ -147,6 +149,38 @@ import org.testng.annotations.Test;
@Test(groups = "broker")
public class BrokerServiceTest extends BrokerTestBase {
+ @Test
+ public void testTopicLoadTimeoutReason() {
+ TopicLoadingContext context = new TopicLoadingContext(TopicName.get("persistent://public/default/test"), true,
+ new CompletableFuture<>());
+ assertEquals(context.getTopicLoadTimeoutReason(), TopicLoadFailureReason.TIMEOUT);
+
+ assertTimeoutReason(context, TopicLoadingStage.NAMESPACE_POLICIES,
+ TopicLoadFailureReason.TIMEOUT_LOAD_NAMESPACE_POLICIES);
+ assertTimeoutReason(context, TopicLoadingStage.TOPIC_POLICIES,
+ TopicLoadFailureReason.TIMEOUT_LOAD_TOPIC_POLICIES);
+ assertTimeoutReason(context, TopicLoadingStage.OPEN_ML, TopicLoadFailureReason.TIMEOUT_LOAD_ML);
+ assertTimeoutReason(context, TopicLoadingStage.INITIALIZE, TopicLoadFailureReason.TIMEOUT_INIT);
+ assertTimeoutReason(context, TopicLoadingStage.PRE_CREATE_COMPACTED_SUB, TopicLoadFailureReason.TIMEOUT_INIT);
+ assertTimeoutReason(context, TopicLoadingStage.REPLICATION, TopicLoadFailureReason.TIMEOUT_INIT);
+ assertTimeoutReason(context, TopicLoadingStage.DEDUPLICATION, TopicLoadFailureReason.TIMEOUT_DEDUP);
+
+ context.start(TopicLoadingStage.INITIALIZE);
+ assertTimeoutReason(context, TopicLoadingStage.NAMESPACE_POLICIES,
+ TopicLoadFailureReason.TIMEOUT_LOAD_NAMESPACE_POLICIES);
+ assertTimeoutReason(context, TopicLoadingStage.TOPIC_POLICIES,
+ TopicLoadFailureReason.TIMEOUT_LOAD_TOPIC_POLICIES);
+ context.finish(TopicLoadingStage.INITIALIZE);
+ }
+
+ private void assertTimeoutReason(TopicLoadingContext context, TopicLoadingStage stage,
+ TopicLoadFailureReason expected) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ context.trace(stage, future);
+ assertEquals(context.getTopicLoadTimeoutReason(), expected);
+ future.complete(null);
+ }
+
@BeforeClass
@Override
protected void setup() throws Exception {
@@ -1202,6 +1236,51 @@ public class BrokerServiceTest extends BrokerTestBase {
}
}
+ @Test
+ public void testTopicLoadMetricsWhenOwnershipCheckFails() throws Exception {
+ String namespace = "prop/disabledBundleMetrics" + UUID.randomUUID().toString().replace("-", "");
+ String topic = "persistent://" + namespace + "/topic";
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("test"), false);
+
+ TopicName topicName = TopicName.get(topic);
+ NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(topicName);
+ pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, false).join();
+
+ JerseyClient httpClient = JerseyClientBuilder.createClient();
+ try {
+ double bundleUnloadingBefore = getTopicLoadFailureMetric(httpClient, "bundle_unloading");
+ double ownershipCheckBefore = getTopicLoadFailureMetric(httpClient, "failed_check_ownership");
+
+ CompletableFuture<Optional<Topic>> topicFuture = pulsar.getBrokerService().getTopic(topicName, true, null);
+ try {
+ topicFuture.get();
+ fail("Topic creation should fail when the namespace bundle is disabled");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException);
+ }
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(getTopicLoadFailureMetric(httpClient, "bundle_unloading"), bundleUnloadingBefore + 1);
+ assertEquals(getTopicLoadFailureMetric(httpClient, "failed_check_ownership"), ownershipCheckBefore);
+ });
+ } finally {
+ httpClient.close();
+ pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, true).join();
+ admin.namespaces().deleteNamespace(namespace);
+ }
+ }
+
+ private double getTopicLoadFailureMetric(JerseyClient httpClient, String reason) {
+ String response = httpClient.target(pulsar.getWebServiceAddress()).path("/metrics/")
+ .request().get(String.class);
+ Multimap<String, PrometheusMetricsClient.Metric> metrics = PrometheusMetricsClient.parseMetrics(response);
+ return metrics.get("topic_load_failed_total").stream()
+ .filter(metric -> reason.equals(metric.tags.get("reason")))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Missing topic load failure metric for " + reason)).value;
+ }
+
@Test
public void testConcurrentLoadTopicExceedLimitShouldNotBeAutoCreated() throws Exception {
boolean needDeleteTopic = false;
@@ -1939,7 +2018,9 @@ public class BrokerServiceTest extends BrokerTestBase {
for (PrometheusMetricsClient.Metric metric : metricMap.get("pulsar_topic_load_failed_count")) {
topicLoadFailedCount += metric.value;
}
- return topicLoadFailedCount >= 1D;
+ return topicLoadFailedCount >= 1D
+ && metricMap.get("topic_load_failed_total").stream()
+ .anyMatch(metric -> "others".equals(metric.tags.get("reason")) && metric.value >= 1D);
});
// Remove the injection.
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 4a248935a387..114fd9b10797 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
@@ -485,8 +485,11 @@ public class PrometheusMetricsTest extends BrokerTestBase {
assertEquals(cm.get(0).tags.get("cluster"), "test");
cm = new ArrayList<>(metrics.get("topic_load_failed_total"));
- assertEquals(cm.size(), 1);
- assertEquals(cm.get(0).tags.get("cluster"), "test");
+ assertEquals(cm.size(), 14);
+ cm.forEach(metric -> {
+ assertEquals(metric.tags.get("cluster"), "test");
+ assertTrue(metric.tags.containsKey("reason"));
+ });
cm = new ArrayList<>(metrics.get("pulsar_in_bytes_total"));
assertEquals(cm.size(), 2);