[Metadata] Fix zk callback thread

[nodece]

Signed-off-by: Zixuan Liu nodeces@gmail.com

Fixes #13211

Motivation

When a method calls the zk metadata store API, this method will use zk callback thread to do some things, once use CompletableFuture#get() or CompletableFuture#join() is used to get a value synchronously in this method, this will cause the zk callback thread into the wait state, when we call the zk metadata store API again, the zk doesn't call the callback because the zk callback is wait state. At the same time, the Pulsar will be shutdown because the session watcher cannot detect whether zk is working.

The jstack log:

"main-EventThread" prio=0 tid=0x0 nid=0x0 waiting on condition
     java.lang.Thread.State: WAITING
 on java.util.concurrent.CompletableFuture$Signaller@6b36f41b
	at java.base@11.0.12/jdk.internal.misc.Unsafe.park(Native Method)
	at java.base@11.0.12/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1796)
	at java.base@11.0.12/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3128)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1823)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture.join(CompletableFuture.java:2043)
	at app//org.apache.pulsar.broker.web.PulsarWebResource.validateTopicOwnership(PulsarWebResource.java:605)
	at app//org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.internalTriggerCompactionNonPartitionedTopic(PersistentTopicsBase.java:3627)
	at app//org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.lambda$internalTriggerCompaction$171(PersistentTopicsBase.java:3602)
	at app//org.apache.pulsar.broker.admin.impl.PersistentTopicsBase$$Lambda$930/0x00000008009f0440.accept(Unknown Source)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture$UniAccept.tryFire$$$capture(CompletableFuture.java:714)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base@11.0.12/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
	at app//org.apache.pulsar.metadata.impl.ZKMetadataStore.handleGetResult(ZKMetadataStore.java:227)
	at app//org.apache.pulsar.metadata.impl.ZKMetadataStore.lambda$batchOperation$6(ZKMetadataStore.java:177)
	at app//org.apache.pulsar.metadata.impl.ZKMetadataStore$$Lambda$479/0x00000008006cb440.processResult(Unknown Source)
	at app//org.apache.pulsar.metadata.impl.PulsarZooKeeperClient$3$1.processResult(PulsarZooKeeperClient.java:490)
	at app//org.apache.zookeeper.ClientCnxn$EventThread.processEvent(ClientCnxn.java:722)
	at app//org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:563)

The Pulsar@2. 9.x has been affected, but I think we also need to cherry-pick this commit to Pulsar@2.8.x

Modifications

• Use Executors.newScheduledThreadPool instead of newSingleThreadExecutor as the metadata store executor
• Fix the ZKMetadataStore#batchOperation() callback thread, use the metadata store executor to execute the callback

Documentation

Need to update docs?

• no-need-doc

[merlimat]

I think we should rather fix this by switching in a different thread in the batch callback:

execute( () -> {
// Trigger all the futures in the batch
                for (int i = 0; i < ops.size(); i++) {
                    OpResult opr = results.get(i);
                    MetadataOp op = ops.get(i);

                    switch (op.getType()) {
                        case PUT:
                            handlePutResult(op.asPut(), opr);
                            break;
                        case DELETE:
                            handleDeleteResult(op.asDelete(), opr);
                            break;
                        case GET:
                            handleGetResult(op.asGet(), opr);
                            break;
                        case GET_CHILDREN:
                            handleGetChildrenResult(op.asGetChildren(), opr);
                            break;

                        default:
                            op.getFuture().completeExceptionally(new MetadataStoreException(
                                    "Operation type not supported in multi: " + op.getType()));
                    }
                }
}

[nodece]

/pulsarbot rerun-failure-checks

[nodece]

/pulsarbot rerun-failure-checks

[nodece]

/pulsarbot rerun-failure-checks

[Jason918]

Can you explains more about why this listenerExecutor is added?

[nodece]

We need to use a single thread to notify listeners.

[nodece]

The current solution looks a little violent, another way is to avoid using future#get() and future#join() anywhere.