[fix][broker] Fix incorrect topic loading latency metric and timeout might not be respected

[BewareMyPower]

Motivation

The topic loading timeout starts at loadOrCreatePersistentTopic:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Lines 1661 to 1663 in 46a76e9

	final CompletableFuture<Optional<Topic>> topicFuture = FutureUtil.createFutureWithTimeout(
	Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(),
	() -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION);

However, there could be two asynchronous tasks before that:

1. Check if the topic exists

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Line 1165 in 46a76e9

return checkNonPartitionedTopicExists(topicName).thenCompose(exists -> {

2. Get the topic policies (__change_events topic will be replayed)

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Line 1171 in 46a76e9

return getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY)

The 1st task should be okay in most time, but the 2nd task could be time-consuming and risky. I have seen when replaying __change_events topic was stuck in production env. If this case happens, there is no metric or log and it can be only aware from client side.

P.S. as the comment shows, it actually does not make sense to execute the 2nd task here but I want to improve the topic loading process in other PRs.

The topic loading metric (pulsar_topic_load_times) starts even later at createPersistent0:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Line 1769 in 46a76e9

final long topicCreateTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());

It does not even include the time when the topic loading is in BrokerService#pendingTopicLoadingQueue, which happens when the number of pending topic loading tasks exceeds the maxConcurrentTopicLoadRequest config.

Modifications

• Refactor TopicLoadingContext to add latency info and create this context from the beginning before checkNonPartitionedTopicExists, so the latency could include the time before createPersistent0
• Log latency when the topic is loaded successfully or fails to load, specially, if the topic has been added to the pending queue due to maxConcurrentTopicLoadRequest limitation, log the pending time as well

Verifying this change

• Make sure that the change passes the CI checks.

Here are example outputs from BrokerServiceTest.testConcurrentLoadTopicExceedLimitShouldNotBeAutoCreated

Created topic persistent://prop/concurrentLoad/my-topic_0 - dedup is disabled (latency: 156 ms)
Created topic persistent://prop/concurrentLoad/my-topic_1 - dedup is disabled (latency: 20 ms)
Created topic persistent://prop/concurrentLoad/my-topic_2 - dedup is disabled (latency: 20 ms)
Created topic persistent://prop/concurrentLoad/my-topic_0 - dedup is disabled (latency: 4 ms)
Created topic persistent://prop/concurrentLoad/my-topic_1 - dedup is disabled (latency: 7 ms (queued: 3))
Created topic persistent://prop/concurrentLoad/my-topic_2 - dedup is disabled (latency: 12 ms (queued: 7))

queued means how long it stays in pendingTopicLoadingQueue.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.20%. Comparing base (1050f48) to head (8e50c15).
⚠️ Report is 398 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #24785       +/-   ##
=============================================
+ Coverage     38.67%   74.20%   +35.53%     
- Complexity    13237    33387    +20150     
=============================================
  Files          1850     1912       +62     
  Lines        144540   149072     +4532     
  Branches      16763    17299      +536     
=============================================
+ Hits          55895   110620    +54725     
+ Misses        81156    29608    -51548     
- Partials       7489     8844     +1355     

Flag	Coverage Δ
inttests	26.60% <83.58%> (-0.09%)	⬇️
systests	22.75% <73.13%> (-0.06%)	⬇️
unittests	73.73% <100.00%> (+38.86%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...rg/apache/pulsar/broker/service/BrokerService.java	83.28% <100.00%> (+25.36%)	⬆️
...che/pulsar/broker/service/TopicLoadingContext.java	100.00% <100.00%> (ø)

... and 1410 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[lhotari]

@BewareMyPower There are some merge conflicts when cherry-picking this to branch-4.0. Could you please handle cherry-picking to branch-4.0? Perhaps there are some dependent PRs that need to be cherry-pick or there's a need to backport.

[BewareMyPower]

Sure, I will do that

[BewareMyPower]

The cherry-picking is done. But when I cherry-picked this PR, I found a regression of this PR that if the topic policies failed to get, the topic loading future won't be complete. I will push a fix soon.