[fix][metadata] Handle session events in a separate thread

[lhotari]

Motivation

• session listener might block further processing, and therefore it's better to run it in another thread
• here's an example of a stacktrace happening in a Bookie with the Pulsar metadata store classes:
  • mentioned in another issue discussion.

metadata-store-zk-session-watcher-21-1  Waiting Thread ID: 60
  jdk.internal.misc.Unsafe.park(Unsafe.java)
  java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)
  java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1796)
  java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3128)
  java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1823)
  java.util.concurrent.CompletableFuture.join(CompletableFuture.java:2043)
  org.apache.pulsar.metadata.coordination.impl.LeaderElectionImpl.close(LeaderElectionImpl.java:233)
  org.apache.pulsar.metadata.bookkeeper.PulsarLedgerAuditorManager.close(PulsarLedgerAuditorManager.java:96)
  org.apache.bookkeeper.replication.AuditorElector.shutdown(AuditorElector.java:234)
  org.apache.bookkeeper.replication.AutoRecoveryMain.shutdown(AutoRecoveryMain.java:154)
  org.apache.bookkeeper.replication.AutoRecoveryMain.lambda$new$0(AutoRecoveryMain.java:99)
  org.apache.bookkeeper.replication.AutoRecoveryMain$$Lambda$222.onSessionExpired()
  org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver.lambda$setSessionStateListener$0(PulsarMetadataClientDriver.java:68)
  org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver$$Lambda$223.accept()
  org.apache.pulsar.metadata.impl.AbstractMetadataStore.lambda$receivedSessionEvent$11(AbstractMetadataStore.java:297)
  org.apache.pulsar.metadata.impl.AbstractMetadataStore$$Lambda$342.accept()
  java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:807)
  org.apache.pulsar.metadata.impl.AbstractMetadataStore.receivedSessionEvent(AbstractMetadataStore.java:295)
  org.apache.pulsar.metadata.impl.ZKMetadataStore.receivedSessionEvent(ZKMetadataStore.java:152)
  org.apache.pulsar.metadata.impl.ZKMetadataStore$$Lambda$148.accept()
  org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkState(ZKSessionWatcher.java:150)
  org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkConnectionStatus(ZKSessionWatcher.java:110)
  org.apache.pulsar.metadata.impl.ZKSessionWatcher$$Lambda$149.run()
  org.apache.pulsar.common.util.Runnables$CatchingAndLoggingRunnable.run(Runnables.java:54)
  java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
  java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
  java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
  java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
  java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
  io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
  java.lang.Thread.run(Thread.java:834)

Modifications

• handle session events in a separate thread

Documentation

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[eolivelli]

One concern that I have with this patch is that usually in ZooKeeper client applications you leverage the feature that all the events are dispatched in strict order.
Here we are moving the execution of the listeners to another thread, and it actually works as the executor is still single threaded.
But on the other hand.. aren't we moving the problem to the metadata-store ?

maybe the problem is to be fixed in AutoRecoveryMain.... onSessionExpired() that is performing a blocking operation (depending on a sync ZooKeeper calll) on the main zookeeper thread.

[lhotari]

Good point. However, in AbstractMetadataStore, the other notifications are delivered using a separate thread pool:

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

Lines 273 to 289 in d8c39bc

	protected CompletableFuture<Void> receivedNotification(Notification notification) {
	try {
	return CompletableFuture.supplyAsync(() -> {
	listeners.forEach(listener -> {
	try {
	listener.accept(notification);
	} catch (Throwable t) {
	log.error("Failed to process metadata store notification", t);
	}
	});
	return null;
	}, executor);
	} catch (RejectedExecutionException e) {
	return FutureUtil.failedFuture(e);
	}
	}

This PR makes AbstractMetadataStore more consistent by also delivering session events using this same thread pool.

[eolivelli]

I noticed that call as well ;-)

So overall I agree with this patch, because it makes it consistent the way we dispatch the events.

But I am not 100% sure that it will fix the original problem

[eolivelli]

+1

but we have to fix the problem also in BookKeeper

[eolivelli]

@lhotari @merlimat
BookKeeper fix apache/bookkeeper#3487

[Jason918]

LGTM