Skip to content

[improve][broker] Trace the asynchronous tasks in logs when loading topics #26163

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BewareMyPower wants to merge 8 commits into
base: master
Choose a base branch
from
+284 −84
Open

[improve][broker] Trace the asynchronous tasks in logs when loading topics #26163

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b8c1805
[improve][broker] Trace the asynchronous tasks in logs when loading t…
BewareMyPower Jul 8, 2026
dbb036e
improve LatencyTracer implementation
BewareMyPower Jul 8, 2026
7ed63b7
move to pulsar-common
BewareMyPower Jul 8, 2026
d98b1b4
make latency tracer pluggable and not require an explicit markEnd tra…
BewareMyPower Jul 13, 2026
ccbcfdf
Merge branch 'master' into bewaremypower/topic-load-tracing
BewareMyPower Jul 13, 2026
9f15124
change some logs to use latencyString
BewareMyPower Jul 13, 2026
de9e6a7
simplify the implementation and remove confusing semantic of setting …
BewareMyPower Jul 13, 2026
b32c02e
improve logging
BewareMyPower Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 51 additions & 60 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -573,7 +573,7 @@
bootstrap.option(ChannelOption.SO_REUSEADDR, true);
bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT);
bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,

Check warning on line 576 in pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

View workflow job for this annotation

GitHub Actions / Build and License check 

[deprecation] RCVBUF_ALLOCATOR in ChannelOption has been deprecated
new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
EventLoopUtil.enableTriggeredMode(bootstrap);
DefaultThreadFactory defaultThreadFactory =
Expand Down Expand Up @@ -601,7 +601,7 @@
bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT);
bootstrap.group(acceptorGroup, workerGroup);
bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,

Check warning on line 604 in pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

View workflow job for this annotation

GitHub Actions / Build and License check 

[deprecation] RCVBUF_ALLOCATOR in ChannelOption has been deprecated
new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
EventLoopUtil.enableTriggeredMode(bootstrap);
Expand Down Expand Up @@ -1305,39 +1305,41 @@
return FutureUtil.failedFuture(new NotAllowedException(
"Broker is unable to load persistent topic"));
}
final var timeoutSeconds = pulsar.getConfiguration().getTopicLoadTimeoutSeconds();
final CompletableFuture<Optional<Topic>> topicFuture = FutureUtil.createFutureWithTimeout(
Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(),
Duration.ofSeconds(timeoutSeconds), executor(),
() -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION);
final var context = new TopicLoadingContext(topicName, createIfMissing, topicFuture);
if (properties != null) {
context.setProperties(properties);
}
topicFuture.exceptionally(t -> {
final var now = System.nanoTime();
final var latency = context.traceAndGetLatency("fail").description();
if (FutureUtil.unwrapCompletionException(t) instanceof TimeoutException) {
log.warn()
.attr("topic", topicName)
.attr("latencyMs", context.latencyMs(now))
.log("Failed to load topic");
.attr("latency", latency)
.log("Failed to load topic within " + timeoutSeconds + " s");
} else {
log.warn()
.attr("topic", topicName)
.attr("latencyMs", context.latencyString(now))
.attr("latency", latency)
.exception(t)
.log("Failed to load topic");
}
pulsarStats.recordTopicLoadFailed();
return Optional.empty();
});
checkNonPartitionedTopicExists(topicName).thenAccept(exists -> {
context.trace("topic exists", checkNonPartitionedTopicExists(topicName)).thenAccept(exists -> {
Comment thread
BewareMyPower marked this conversation as resolved.
Comment thread
BewareMyPower marked this conversation as resolved.
if (!exists && !createIfMissing) {
topicFuture.complete(Optional.empty());
return;
}
// The topic level policies are not needed now, but the meaning of calling
// "getTopicPoliciesBypassSystemTopic" will wait for system topic policies initialization.
getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY)
.thenRun(() -> {
final var systemTopicLoadFuture = context.trace("system topic",
getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY));
systemTopicLoadFuture.thenRun(() -> {
final var inserted = new MutableBoolean(false);
final var cachedFuture = topics.computeIfAbsent(topicName.toString(), ___ -> {
inserted.setTrue();
Expand All @@ -1349,7 +1351,7 @@
// actual loading latency that should not be recorded in metrics.
log.info()
.attr("topic", topicName)
.attr("latencyMs", context.latencyString(System.nanoTime()))
.attr("latency", context.getLatency().description())
.log("Finished loading from other concurrent loading task");
cachedFuture.whenComplete((optTopic, e) -> {
if (e == null) {
Expand All @@ -1360,15 +1362,14 @@
});
}
}).exceptionally(e -> {
pulsar.getExecutor().execute(() -> topics.remove(topicName.toString(), topicFuture));
final Throwable rc = FutureUtil.unwrapCompletionException(e);
log.error().attr("topic", topicName)
.exceptionMessage(rc)
.log("Topic creation encountered an exception"
+ " by initialize topic policies service");
topicFuture.completeExceptionally(rc);
log.warn().attr("topic", topicName).log("Topic creation encountered an exception"
+ " by initialize topic policies service");
failTopicFuture(topicName.toString(), topicFuture, e);
return null;
});
}).exceptionally(e -> {
failTopicFuture(topicName.toString(), topicFuture, e);
return null;
});
return topicFuture;
} else {
Expand Down Expand Up @@ -1420,6 +1421,14 @@
}
}

private void failTopicFuture(String topic, CompletableFuture<Optional<Topic>> topicFuture, Throwable throwable) {
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
final Throwable rc = FutureUtil.unwrapCompletionException(throwable);
// It will trigger the logging for exception and traced latencies in topicFuture's exceptionally callback, so
// we don't need to add an extra log before it.
topicFuture.completeExceptionally(rc);
}

private CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesBypassSystemTopic(@NonNull TopicName topicName,
TopicPoliciesService.GetType type) {
if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
Expand Down Expand Up @@ -1859,7 +1868,7 @@
protected CompletableFuture<Optional<Topic>> loadOrCreatePersistentTopic(TopicLoadingContext context) {
final var topic = context.getTopicName().toString();
final var topicFuture = context.getTopicFuture();
checkTopicNsOwnership(topic)
context.trace("ownership", checkTopicNsOwnership(topic))
.thenRun(() -> {
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();

Expand All @@ -1871,8 +1880,7 @@
// do not recreate topic if topic is already migrated and deleted by broker
// so, avoid creating a new topic if migration is already started
if (ex != null && (ex.getCause() instanceof TopicMigratedException)) {
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(ex.getCause());
failTopicFuture(topic, topicFuture, ex);
return null;
}
createPendingLoadTopic();
Expand All @@ -1883,8 +1891,7 @@
log.debug().attr("topic", topic).log("topic-loading for added into pending queue");
}
}).exceptionally(ex -> {
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(ex.getCause());
failTopicFuture(topic, topicFuture, ex);
return null;
});

Expand Down Expand Up @@ -1924,27 +1931,24 @@
TopicName topicName = context.getTopicName();
final var topic = topicName.toString();
final var topicFuture = context.getTopicFuture();
checkTopicNsOwnership(topic).thenRun(() -> {
context.trace("2nd ownership", checkTopicNsOwnership(topic)).thenRun(() -> {
CompletableFuture<Map<String, String>> propertiesFuture;
if (context.getProperties() == null) {
//Read properties from storage when loading topic.
propertiesFuture = fetchTopicPropertiesAsync(topicName);
} else {
propertiesFuture = CompletableFuture.completedFuture(context.getProperties());
}
propertiesFuture.thenAccept(finalProperties -> {
context.trace("properties", propertiesFuture).thenAccept(finalProperties -> {
context.setProperties(finalProperties);
//TODO add topicName in properties?
createPersistentTopic0(context);
}).exceptionally(throwable -> {
log.warn().attr("topic", topic).exception(throwable).log("Read topic property failed");
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(throwable);
failTopicFuture(topic, topicFuture, throwable);
return null;
});
}).exceptionally(e -> {
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(e.getCause());
failTopicFuture(topic, topicFuture, e);
return null;
});
}
Expand All @@ -1957,10 +1961,8 @@
final var createIfMissing = context.isCreateIfMissing();

if (isTransactionInternalName(topicName)) {
String msg = String.format("Can not create transaction system topic %s", topic);
log.warn(msg);
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(new NotAllowedException(msg));
failTopicFuture(topic, topicFuture, new NotAllowedException("Can not create transaction system topic "
+ topic));
return;
}

Expand All @@ -1977,6 +1979,7 @@
n.recycle();
return found;
}), (managedLedgerConfig, exists) -> {
context.trace("ml-config");
if (isBrokerEntryMetadataEnabled() || isBrokerPayloadProcessorEnabled()) {
// init managedLedger interceptor
Set<BrokerEntryMetadataInterceptor> interceptors = new HashSet<>();
Expand Down Expand Up @@ -2038,27 +2041,26 @@
@Override
public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
try {
context.trace("open-ml");
PersistentTopic persistentTopic = isSystemTopic(topic)
? new SystemTopic(topic, ledger, BrokerService.this)
: newTopic(topic, ledger, BrokerService.this, PersistentTopic.class);
persistentTopic.setCreateFuture(topicFuture);
persistentTopic
.initialize()
.thenCompose(__ -> persistentTopic.preCreateSubscriptionForCompactionIfNeeded())
.thenCompose(__ -> persistentTopic.checkReplication())
.thenCompose(v -> {
// Also check dedup status
return persistentTopic.checkDeduplicationStatus();
})
context.trace("init", persistentTopic.initialize())
.thenCompose(__ -> context.trace("pre-create compacted sub",
persistentTopic.preCreateSubscriptionForCompactionIfNeeded()))
.thenCompose(__ -> context.trace("replication",
persistentTopic.checkReplication()))
.thenCompose(v -> context.trace("deduplication",
persistentTopic.checkDeduplicationStatus()))
.thenRun(() -> {
long nowInNanos = System.nanoTime();
long topicLoadLatencyMs = context.latencyMs(nowInNanos);
final var latency = context.traceAndGetLatency("done");
log.info()
.attr("topic", topic)
.attr("dedupEnabled", persistentTopic.isDeduplicationEnabled())
.attr("latencyMs", context.latencyString(nowInNanos))
.attr("latency", latency.description())
.log("Loaded topic");
pulsarStats.recordTopicLoadTimeValue(topic, topicLoadLatencyMs);
pulsarStats.recordTopicLoadTimeValue(topic, latency.elapsedInMillis());
if (!topicFuture.complete(Optional.of(persistentTopic))) {
// Check create persistent topic timeout.
if (topicFuture.isCompletedExceptionally()) {
Expand Down Expand Up @@ -2110,12 +2112,7 @@
return null;
});
} catch (Exception e) {
log.warn()
.attr("topic", topic)
.exceptionMessage(e)
.log("Failed to create topic");
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(e);
failTopicFuture(topic, topicFuture, e);
}
}

Expand All @@ -2127,9 +2124,7 @@
loadFuture.completeExceptionally(exception);
topicFuture.complete(Optional.empty());
} else {
log.warn().attr("topic", topic).exception(exception).log("Failed to create topic");
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(new PersistenceException(exception));
failTopicFuture(topic, topicFuture, new PersistenceException(exception));
}
}
}, () -> isTopicNsOwnedByBrokerAsync(topicName), null);
Expand All @@ -2140,12 +2135,8 @@
"Failed to get topic configuration";
log.warn()
.attr("topic", topic)
.exception(exception)
.log(msg);
// remove topic from topics-map in different thread to avoid possible deadlock if
// createPersistentTopic-thread only tries to handle this future-result
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
topicFuture.completeExceptionally(exception);
failTopicFuture(topic, topicFuture, exception);
return null;
});
}
Expand Down Expand Up @@ -3571,9 +3562,9 @@
return;
}

pendingTopic.polledFromQueue();
pendingTopic.trace("queued");
final String topic = pendingTopic.getTopicName().toString();
checkTopicNsOwnership(topic).thenRun(() -> {
pendingTopic.trace("ownership", checkTopicNsOwnership(topic)).thenRun(() -> {
CompletableFuture<Optional<Topic>> pendingFuture = pendingTopic.getTopicFuture();
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
final boolean acquiredPermit = topicLoadSemaphore.tryAcquire();
Expand Down
34 changes: 10 additions & 24 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,15 @@
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentLinkedQueue;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.LatencyTracer;
import org.jspecify.annotations.Nullable;

@RequiredArgsConstructor
public class TopicLoadingContext {
public class TopicLoadingContext extends LatencyTracer {

private static final String EXAMPLE_LATENCY_OUTPUTS = "1234 ms (queued: 567)";

private final long startNs = System.nanoTime();
@Getter
private final TopicName topicName;
@Getter
Expand All @@ -43,23 +39,13 @@ public class TopicLoadingContext {
@Getter
@Setter
@Nullable private Map<String, String> properties;
private long polledFromQueueNs = -1L;

public void polledFromQueue() {
polledFromQueueNs = System.nanoTime();
}

public long latencyMs(long nowInNanos) {
return TimeUnit.NANOSECONDS.toMillis(nowInNanos - startNs);
}

public String latencyString(long nowInNanos) {
final var builder = new StringBuilder(EXAMPLE_LATENCY_OUTPUTS.length());
builder.append(latencyMs(nowInNanos));
builder.append(" ms");
if (polledFromQueueNs >= 0) {
builder.append(" (queued: ").append(latencyMs(polledFromQueueNs)).append(")");
}
return builder.toString();
public TopicLoadingContext(TopicName topicName, boolean createIfMissing,
CompletableFuture<Optional<Topic>> topicFuture) {
// The topic loading could be ended asynchronously by a timeout event, so we need a thread safe queue here
super(new ConcurrentLinkedQueue<>(), System::nanoTime);
this.topicName = topicName;
this.createIfMissing = createIfMissing;
this.topicFuture = topicFuture;
}
}
Loading
Loading