[improve][broker] Optimize and clean up aggregation of topic stats

[lhotari]

Motivation

• getPublishers method gets called twice for every single publisher when aggregating topic stats in the add method of TopicStatsImpl:

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java

Lines 262 to 263 in 1a352f1

	for (int index = 0; index < stats.getPublishers().size(); index++) {
	PublisherStats s = stats.getPublishers().get(index);

This is a slight problem since getPublishers method is fairly costly and causes unnecessary garbage:

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java

Lines 152 to 158 in 1a352f1

	public List<? extends PublisherStats> getPublishers() {
	return Stream.concat(publishers.stream().sorted(
	Comparator.comparing(PublisherStatsImpl::getProducerName, nullsLast(naturalOrder()))),
	publishersMap.values().stream().sorted(
	Comparator.comparing(PublisherStatsImpl::getProducerName, nullsLast(naturalOrder()))))
	.collect(Collectors.toList());
	}

• The code for aggregating subscription and replication stats is unnecessarily complex.

Modifications

• extract a variable before the loop to eliminate the problem mentioned above.
• refactor the code for aggregating subscription and replication stats

Documentation

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

[heesung-sn]

Can we also cache the sorted publisher list? It seems unnecessary to sort the same list repeatedly.

[codecov-commenter]

Codecov Report

Merging #21361 (49d31a8) into master (1a352f1) will increase coverage by 39.83%.
Report is 2 commits behind head on master.
The diff coverage is 80.00%.

@@              Coverage Diff              @@
##             master   #21361       +/-   ##
=============================================
+ Coverage     33.49%   73.32%   +39.83%     
- Complexity    12257    32553    +20296     
=============================================
  Files          1636     1888      +252     
  Lines        127789   140255    +12466     
  Branches      13963    15435     +1472     
=============================================
+ Hits          42798   102846    +60048     
+ Misses        79372    29324    -50048     
- Partials       5619     8085     +2466     

Flag	Coverage Δ
inttests	24.28% <0.00%> (+0.13%)	⬆️
systests	24.70% <0.00%> (?)
unittests	72.62% <80.00%> (+40.62%)	⬆️

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

Files	Coverage Δ
...sar/common/policies/data/stats/TopicStatsImpl.java	94.70% <100.00%> (+41.90%)	⬆️
...licies/data/stats/NonPersistentTopicStatsImpl.java	83.78% <62.50%> (+72.54%)	⬆️

... and 1533 files with indirect coverage changes

[lhotari]

Can we also cache the sorted publisher list? It seems unnecessary to sort the same list repeatedly.

One possibility is to eliminate the getPublishers method. It looks like it doesn't make much sense at the moment. That could be refactored in another PR.