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

[BewareMyPower]

Motivation

The topic loading time could be high with pressure, but it's hard to measure which sub-task takes the most time. This is a follow-up work of #24785 to improve the logs that show topic loading latency.

Modifications

Add a simple tracing class LatencyTracer and trace asynchronous futures during topic loading. When tracing a future, if the future is already done, skip tracing it because such tasks are usually not CPU-bounded.

Verifying this change

LatencyTracerTest covers functionality of all public methods with injected timestamps. The topic loading logs can be checked by running testConcurrentLoadTopicExceedLimitShouldNotBeAutoCreated:

Loaded topic {dedupEnabled=false, latency=total: 122 ms, ml-config: 9 ms, open-ml: 58 ms, init: 53 ms, done: 870 us, topic=persistent://prop/concurrentLoad/my-topic_0}
Loaded topic {dedupEnabled=false, latency=total: 19 ms, ml-config: 936 us, open-ml: 17 ms, init: 1 ms, done: 110 us, topic=persistent://prop/concurrentLoad/my-topic_1}
Loaded topic {dedupEnabled=false, latency=total: 20 ms, ml-config: 834 us, open-ml: 18 ms, init: 1 ms, done: 122 us, topic=persistent://prop/concurrentLoad/my-topic_2}
Loaded topic {dedupEnabled=false, latency=total: 4 ms, properties: 2 ms, ml-config: 911 us, open-ml: 89 us, init: 642 us, done: 91 us, topic=persistent://prop/concurrentLoad/my-topic_0}
Loaded topic {dedupEnabled=false, latency=total: 7 ms, queued: 2 ms, properties: 2 ms, ml-config: 692 us, open-ml: 88 us, init: 788 us, done: 147 us, topic=persistent://prop/concurrentLoad/my-topic_1}
Loaded topic {dedupEnabled=false, latency=total: 12 ms, queued: 7 ms, properties: 3 ms, ml-config: 907 us, open-ml: 93 us, init: 634 us, done: 81 us, topic=persistent://prop/concurrentLoad/my-topic_2}

Out of scope

In my previous tests, opening managed ledger could also take much time, this LatencyTracer class can be used to track managed ledger open process as well.

[void-ptr974]

Thanks for the improvement! The additional timing details should help with understanding topic-loading latency. I left a couple of minor comments.

[BewareMyPower]

Hi, @Denovo1998 @void-ptr974 I've addressed your comments in latest commit with a few improvements, PTAL.

1. Before that, the tracer requires users to explicitly call trace before calling latencyInMillis() or latencyString(), which is not friendly, so the latest design adopts an efficient CAS call to ensure the last timestamp could be computed lazily with "done" as the stage description.
2. For thread safety issue, I think it's caused by a bad design of the topic loading timeout mechanism. But I'd like to avoid touching that issue in this PR, so I will simply change TopicLoadingContext's timepoints to a thread safe queue. If the timeout mechanism can be removed in future, it can be changed back to a not thread safe queue like LinkedList.

[Denovo1998]

Perhaps the implementation can be simpler?

Each time a log or metric needs to be output, call snapshot() once, take the current time as endNs within it, collect only the checkpoints not later than endNs, and then return an immutable LatencySnapshot. Both logs and metrics take values from the same snapshot.

This approach eliminates the need for intermediate states like -1/-2, and reading latency will not inadvertently permanently end the tracer. Even if timeout logs have been printed, the background topic loading can still continue to record subsequent stages.

Additionally, currently, when a timeout occurs, incomplete stages only show something like done: 29999 ms. It's not clear from the logs whether the actual stall is in the system topic, open-ml, or another future. Since the main goal of this PR is to identify which specific stage is consuming time, it's best to record the current action when a future starts and record the end upon completion. This way, the timeout logs can directly indicate at which step the process is stuck.

[BewareMyPower]

I agree that freezing the end time in latencyString or latencyInMillis is not a good idea. I adopted a simplified implementation and completed topicFuture exceptionally when checkNonPartitionedTopicExists fails, could you take a look again? @Denovo1998

[Denovo1998]

This path remains incomplete. The new exception handler is attached to systemTopicLoadFuture.thenRun(...) inside the thenAccept body. If checkNonPartitionedTopicExists() fails, thenAccept is skipped, leaving the inner handler unregistered and topicFuture waiting until timeout.

Could we attach the failure handler to the future returned by the outer context.trace(...).thenAccept(...), or flatten the sequence with thenCompose and handle its terminal failure once?

[Denovo1998]

A thread may be paused between acquiring a timestamp and calling add():

• A acquires a 10 ms timestamp, then pauses.

• The timeout thread acquires and inserts fail: 20 ms.

• A resumes, inserts A: 10 ms after fail.

• getLatency() calculates according to the queue order and treats the last element as the total end time.

---

The concurrent queue prevents structural corruption, but it does not preserve timestamp order. trace() samples the clock before Queue.add(), so one thread can be descheduled after sampling and insert an older timepoint after a newer one. getLatency() then trusts queue order and uses the last element as the total end time.

Could trace insertion and snapshot creation be serialized, or could the snapshot capture a terminal timestamp and order/filter the copied timepoints by timestamp? A concurrent regression test would also be useful here.

[Denovo1998]

It feels a bit strange here? This resolves the same Future twice. Since the first CompleteExceptionally(e) wins, the later CompleteExceptionally(rc) is always a no-op. Callers will receive the wrapped exception (e) rather than rc.

Completing it once with rc will still trigger the registered failure-tracing callback. Could we unwrap first and call completeExceptionally(rc) only once?

[Denovo1998]

The timeout total is correct now, but the log still reports the unfinished interval as fail: 29999 ms. Since actions are recorded only after their futures complete, it does not directly identify whether system topic, open ml, or another stage is currently stuck.

If identifying the stalled subtask is part of this PR’s goal, could we also record the action when it starts? This should add negligible overhead because only a few actions are traced, though the implementation should account for overlapping actions and avoid clearing a newer action when an older future completes.