[fix][broker] Execute the pending callbacks in order before ready for incoming requests

[BewareMyPower]

Background Knowledge

CompletableFuture has an intuitive behavior

        final var future = new CompletableFuture<Integer>();
        future.thenRun(() -> System.out.println("A"));
        future.thenRun(() -> System.out.println("B"));
        future.thenRun(() -> System.out.println("C"));
        future.complete(0);

The outputs of the code above are:

C
B
A

That's because it maintains callbacks in the LIFO stack, not FIFO queue.

Motivation

#22977 breaks the order of some events during extensible load manager's start by adding the runnable objects via thenRun(). The previous events order of ExtensibleLoadManagerImpl:

1. playLeader() or playFollower()
2. serviceUnitStateChannel.start()
3. Schedule some tasks (brokerLoadDataReportTask, topBundlesLoadDataReportTask, etc.) and set started with true.

Now, since they will be executed in reverse order, started will first be set true and then event 1 and 2 happened. It might cause unexpected issues

Modifications

Add a synchronized list to queue the pending tasks and execute them in order before the future is complete.

Documentation

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

Matching PR in forked repository

PR in forked repository:

[heesung-sn]

Nice catch. I think we also have other pulsar logics, e.g. FutureUtil.sequencer that uses future chain for ordering. I wonder if we need to revisit that too.

[lhotari]

Nice catch. I think we also have other pulsar logics, e.g. FutureUtil.sequencer that uses future chain for ordering. I wonder if we need to revisit that too.

Yes, nice catch @BewareMyPower.

It's useful to consider replacing FutureUtil.sequencer usage with com.spotify.futures.ConcurrencyReducer with the concurrency of 1. ConcurrencyReducer is already included as a dependency in Pulsar as part of com.spotify:completable-futures.

[heesung-sn]

Actually, I see that the sequencer uses thenCompose, which I think the order should be fifo.

[lhotari]

Actually, I see that the sequencer uses thenCompose, which I think the order should be fifo.

I think it's different.

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java

Lines 234 to 242 in 9626e7e

	if (sequencerFuture.isDone()) {
	if (sequencerFuture.isCompletedExceptionally() && allowExceptionBreakChain) {
	return sequencerFuture;
	}
	return sequencerFuture = newTask.get();
	}
	return sequencerFuture = allowExceptionBreakChain
	? sequencerFuture.thenCompose(__ -> newTask.get())
	: sequencerFuture.exceptionally(ex -> null).thenCompose(__ -> newTask.get());

the instances are chained and .thenCompose isn't called multiple times for the same instance.

[BewareMyPower]

The sequencer is different because it's a chained futures (like linked list) that each future has only 1 callback in its internal stack while the existing design is a single future with N callbacks. However, the sequencer might be unnecessarily complicated for this simple case so I won't adopt it.

[BewareMyPower]

I found a deadlock in ClusterMigrationTest. Mark this PR as drafted.

[lhotari]

Good work @BewareMyPower !

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.55%. Comparing base (bbc6224) to head (12b0c60).
Report is 566 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23266      +/-   ##
============================================
+ Coverage     73.57%   74.55%   +0.97%     
- Complexity    32624    33779    +1155     
============================================
  Files          1877     1926      +49     
  Lines        139502   145056    +5554     
  Branches      15299    15864     +565     
============================================
+ Hits         102638   108141    +5503     
+ Misses        28908    28646     -262     
- Partials       7956     8269     +313     

Flag	Coverage Δ
inttests	27.86% <100.00%> (+3.28%)	⬆️
systests	24.67% <100.00%> (+0.34%)	⬆️
unittests	73.90% <100.00%> (+1.06%)	⬆️

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

Files with missing lines	Coverage Δ
...n/java/org/apache/pulsar/broker/PulsarService.java	83.82% <100.00%> (+1.45%)	⬆️

... and 556 files with indirect coverage changes