[fix][broker] Continue closing even when executor is shut down

[lhotari]

Motivation

2024-04-26T09:36:28,397 - ERROR - [ForkJoinPool.commonPool-worker-10:PersistentTopic] - [persistent://pulsar/global/removeClusterTest/__change_events] Error closing clients
java.util.concurrent.CompletionException: java.util.concurrent.RejectedExecutionException: Executor is shutting down
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniRunNow(CompletableFuture.java:823) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniRunStage(CompletableFuture.java:803) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenRunAsync(CompletableFuture.java:2204) ~[?:?]
	at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$delete$44(PersistentTopic.java:1430) ~[classes/:?]
	at org.apache.pulsar.broker.resources.NamespaceResources$PartitionedTopicResources.lambda$runWithMarkDeleteAsync$9(NamespaceResources.java:359) ~[pulsar-broker-common-3.3.0-SNAPSHOT.jar:3.3.0-SNAPSHOT]
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) [?:?]
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) [?:?]
	at java.base/java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:483) [?:?]
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) [?:?]
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) [?:?]
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) [?:?]
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) [?:?]
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) [?:?]
Caused by: java.util.concurrent.RejectedExecutionException: Executor is shutting down
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.execute(SingleThreadExecutor.java:208) ~[bookkeeper-common-4.17.0.jar:4.17.0]
	at java.base/java.util.concurrent.CompletableFuture.uniRunNow(CompletableFuture.java:817) ~[?:?]
	... 12 more

Modifications

• provide custom Executor that catches RejectedExecutionException and runs in current thread in that case.

Documentation

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