Async the method onManagedLedgerLastLedgerInitialize for ManagedLedgerInterceptor

[codelipenghui]

Fixes #10703

Motivation

Async the method onManagedLedgerLastLedgerInitialize for ManagedLedgerInterceptor. The root cause for issue #10703 is the ManagedLedger Interceptor is making a block read call from within a thread used for async operations. The interceptor shouldn’t be making any blocking calls. Thanks for @merlimat's help.

Here is the detailed stacktrace:

"BookKeeperClientWorker-OrderedExecutor-0-0" #67 prio=5 os_prio=0 cpu=50.69ms elapsed=14871.45s allocated=4671K defined_classes=63 tid=0x00007f8cee18e560 nid=0x9b01 waiting on condition  [0x00007f8b8fbfa000]
   java.lang.Thread.State: WAITING (parking)
 at jdk.internal.misc.Unsafe.park(java.base@11.0.11/Native Method)
 - parking to wait for  <0x0000000602818fc8> (a java.util.concurrent.CompletableFuture$Signaller)
 at java.util.concurrent.locks.LockSupport.park(java.base@11.0.11/LockSupport.java:194)
 at java.util.concurrent.CompletableFuture$Signaller.block(java.base@11.0.11/CompletableFuture.java:1796)
 at java.util.concurrent.ForkJoinPool.managedBlock(java.base@11.0.11/ForkJoinPool.java:3128)
 at java.util.concurrent.CompletableFuture.waitingGet(java.base@11.0.11/CompletableFuture.java:1823)
 at java.util.concurrent.CompletableFuture.get(java.base@11.0.11/CompletableFuture.java:1998)
 at org.apache.bookkeeper.common.concurrent.FutureUtils.result(FutureUtils.java:72)
 at org.apache.bookkeeper.client.api.ReadHandle.read(ReadHandle.java:58)
 at org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImpl.onManagedLedgerLastLedgerInitialize(ManagedLedgerInterceptorImpl.java:89)
 at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl$1.lambda$null$0(ManagedLedgerImpl.java:358)
 at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl$1$$Lambda$800/0x00000008406bb040.run(Unknown Source)
 at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32)
 at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@11.0.11/ThreadPoolExecutor.java:1128)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@11.0.11/ThreadPoolExecutor.java:628)
 at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
 at java.lang.Thread.run(java.base@11.0.11/Thread.java:829)

   Locked ownable synchronizers:
 - <0x0000000601a92180> (a java.util.concurrent.ThreadPoolExecutor$Worker)

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

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[merlimat]

We can probably just do the parsing once

[merlimat]

In general it's preferable to do:

Suggested change

	.whenComplete((__, ex) -> {
	if (ex != null) {
	callback.initializeFailed(new ManagedLedgerInterceptException(ex));
	} else {
	initializeBookKeeper(callback);
	}
	});
	.thenRun(() -> initializeBookKeeper(callback))
	.exceptionally(ex -> {
	callback.initializeFailed(new ManagedLedgerInterceptException(ex));
	return null;
	});

The reason is that if there is any unexpected exception thrown when calling initializeBookKeeper(), the exceptionally part will be invoked.

[hangc0276]

Good Job!