[fix][broker] Fix topic policies cannot be queried with extensible load manager

[BewareMyPower]

Motivation

#23319 brings a regression that topic policies cannot be queried with extensible load manager. compactionScheduleTest could not succeed now.

In SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync, it skips getting events from __change_events because the reader will fail to create if the load manager does not start.

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

Lines 253 to 259 in 4b3b273

	// When the extensible load manager initializes its channel topic, it will trigger the topic policies
	// initialization by calling this method. At the moment, the load manager does not start so the lookup
	// for "__change_events" will fail. In this case, just return an empty policies to avoid deadlock.
	final var loadManager = pulsarService.getLoadManager().get();
	if (loadManager == null || !loadManager.started()) {
	return CompletableFuture.completedFuture(Optional.empty());
	}

However, the state validation in ServiceUnitStateChannelImpl here is wrong:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

Line 270 in 4b3b273

return validateChannelState(LeaderElectionServiceStarted, false);

The method above returns true when the state is Closed, Constructed or LeaderElectionServiceStarted, which is the opposite of the expected result (Started).

Modifications

Fix the ServiceUnitStateChannelImpl#started() method with the correct state validation.

However, if only with this simple fix, the extensible load manager would fail to start. It's because after the table view creation for loadbalancer-broker-load-data and loadbalancer-top-bundles-load-data topics happen after the load manager start. In isAllowAutoSubscriptionCreationAsync, it will wait until the future of getTopicPoliciesAsync on these topics is done, while the reader creation on __change_events will fail.

Therefore, to fix this start failure, just return true for internal topics in isAllowAutoSubscriptionCreationAsync.

Verifying this change

ExtensibleLoadManagerImplTest#compactionScheduleTest will pass now. (It's in the flaky test suite)

Documentation

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

Matching PR in forked repository

PR in forked repository:

[BewareMyPower]

ExtensibleLoadManagerCloseTest failed now. Let me take a look

[BewareMyPower]

I skipped getTopicPoliciesAsync calls for extensible LM's internal topics in BrokerService's topic initialization, let's see if all tests can pass now.

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.65%. Comparing base (bbc6224) to head (b601f85).
Report is 591 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23326      +/-   ##
============================================
+ Coverage     73.57%   74.65%   +1.08%     
- Complexity    32624    34409    +1785     
============================================
  Files          1877     1929      +52     
  Lines        139502   145158    +5656     
  Branches      15299    15884     +585     
============================================
+ Hits         102638   108374    +5736     
+ Misses        28908    28530     -378     
- Partials       7956     8254     +298     

Flag	Coverage Δ
inttests	27.91% <100.00%> (+3.32%)	⬆️
systests	24.69% <0.00%> (+0.36%)	⬆️
unittests	74.00% <100.00%> (+1.16%)	⬆️

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

Files with missing lines	Coverage Δ
...xtensions/channel/ServiceUnitStateChannelImpl.java	87.25% <100.00%> (+1.95%)	⬆️
...rg/apache/pulsar/broker/service/BrokerService.java	82.69% <100.00%> (+1.91%)	⬆️
...sar/broker/service/persistent/PersistentTopic.java	79.58% <100.00%> (+1.12%)	⬆️

... and 589 files with indirect coverage changes