[fix][broker] Fixed the ExtensibleLoadManagerImpl internal system getTopic failure when the leadership changes

[heesung-sn]

Motivation

After this fix, #21729, we need to recreate the non-persistent internal system topics when the leadership changes because we disabled the auto-creation of the non-persistent internal system topics.

Also, the non-persistent consumers and producers might be in a closed state, as they could receive TopicNotFoundException when the new leader broker has not created the internal topics yet. To cover this case, we need to update the leader and follower role change logic to be more idempotent.

Modifications

• Make the new leader always try to create internal non-persistent system topics.
• Init the load data stores whenever the leader broker has changed.

Verifying this change

• Make sure that the change passes the CI checks.

ExtensibleLoadManager.testIsolationPolicy test was flaky because of the this issue.

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: heesung-sn#55

[dragosvictor]

LGTM!

[lhotari]

LGTM

[codecov-commenter]

Codecov Report

Attention: 12 lines in your changes are missing coverage. Please review.

Comparison is base (0ad1867) 72.94% compared to head (8b5062f) 73.43%.
Report is 4 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21764      +/-   ##
============================================
+ Coverage     72.94%   73.43%   +0.49%     
- Complexity    32579    32786     +207     
============================================
  Files          1897     1897              
  Lines        140627   140647      +20     
  Branches      15486    15489       +3     
============================================
+ Hits         102581   103288     +707     
+ Misses        29974    29285     -689     
- Partials       8072     8074       +2     

Flag	Coverage Δ
inttests	24.17% <4.54%> (?)
systests	24.94% <4.54%> (+0.23%)	⬆️
unittests	72.71% <81.81%> (+0.38%)	⬆️

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

Files	Coverage Δ
...ervice/AbstractDispatcherSingleActiveConsumer.java	90.55% <100.00%> (+3.24%)	⬆️
...a/org/apache/pulsar/broker/service/Dispatcher.java	55.55% <100.00%> (+11.80%)	⬆️
...tent/NonPersistentDispatcherMultipleConsumers.java	68.67% <100.00%> (+0.38%)	⬆️
...sistent/PersistentDispatcherMultipleConsumers.java	75.54% <ø> (+7.59%)	⬆️
...tent/PersistentDispatcherSingleActiveConsumer.java	71.25% <ø> (+4.68%)	⬆️
...e/extensions/store/TableViewLoadDataStoreImpl.java	80.39% <71.42%> (-9.09%)	⬇️
...dbalance/extensions/ExtensibleLoadManagerImpl.java	79.55% <83.33%> (+0.04%)	⬆️

... and 119 files with indirect coverage changes

[Demogorgon314]

@heesung-sn Can you help cherry-pick this PR to branch-3.1?

[heesung-sn]

Raised a cherry-pick PR here.
#21801