From 9464d68070de31c6d39b4769ac80b5c1773bf8cd Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Tue, 21 Mar 2023 10:13:53 +0800 Subject: [PATCH 1/3] Make session notification to be async. --- .../apache/pulsar/common/util/FutureUtil.java | 10 +++++++++ .../coordination/impl/LeaderElectionImpl.java | 22 ++++++++++--------- .../coordination/impl/LockManagerImpl.java | 20 ++++++++--------- 3 files changed, 32 insertions(+), 20 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java index 531769a1e452b..8cde6e7c99127 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java @@ -36,6 +36,7 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; +import javax.annotation.Nonnull; import javax.annotation.concurrent.ThreadSafe; /** @@ -238,6 +239,15 @@ public static CompletableFuture addTimeoutHandling(CompletableFuture f return future; } + public static @Nonnull CompletableFuture composeAsync(Supplier> futureSupplier, + Executor executor) { + Objects.requireNonNull(futureSupplier); + Objects.requireNonNull(executor); + return CompletableFuture.completedFuture(null) + .thenComposeAsync(__ -> futureSupplier.get(), executor); + } + + /** * Creates a low-overhead timeout exception which is performance optimized to minimize allocations * and cpu consumption. It sets the stacktrace of the exception to the given source class and diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java index c1121b1309c2c..b2ac4b6bb5bb9 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java @@ -32,6 +32,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.common.util.SafeRunnable; +import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataCache; import org.apache.pulsar.metadata.api.MetadataCacheConfig; @@ -62,6 +63,7 @@ class LeaderElectionImpl implements LeaderElection { private Optional proposedValue; private final ScheduledExecutorService executor; + private final FutureUtil.Sequencer sequencer; private enum InternalState { Init, ElectionInProgress, LeaderIsPresent, Closed @@ -85,7 +87,7 @@ private enum InternalState { this.internalState = InternalState.Init; this.stateChangesListener = stateChangesListener; this.executor = executor; - + this.sequencer = FutureUtil.Sequencer.create(); store.registerListener(this::handlePathNotification); store.registerSessionListener(this::handleSessionNotification); updateCachedValueFuture = executor.scheduleWithFixedDelay(SafeRunnable.safeRun(this::getLeaderValue), @@ -277,18 +279,18 @@ public Optional getLeaderValueIfPresent() { private void handleSessionNotification(SessionEvent event) { // Ensure we're only processing one session event at a time. - executor.execute(SafeRunnable.safeRun(() -> { + sequencer.sequential(() -> FutureUtil.composeAsync(() -> { if (event == SessionEvent.SessionReestablished) { log.info("Revalidating leadership for {}", path); - - try { - LeaderElectionState les = elect().get(); - log.info("Resynced leadership for {} - State: {}", path, les); - } catch (ExecutionException | InterruptedException e) { - log.warn("Failure when processing session event", e); - } + return elect().thenAccept(leaderState -> { + log.info("Resynced leadership for {} - State: {}", path, leaderState); + }).exceptionally(ex -> { + log.warn("Failure when processing session event", ex); + return null; + }); } - })); + return CompletableFuture.completedFuture(null); + }, executor)); } private void handlePathNotification(Notification notification) { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java index f5e7e0528bd7b..38ce58b2d3bc1 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java @@ -53,6 +53,7 @@ class LockManagerImpl implements LockManager { private final MetadataStoreExtended store; private final MetadataCache cache; private final MetadataSerde serde; + private final FutureUtil.Sequencer sequencer; private final ExecutorService executor; private enum State { @@ -72,6 +73,7 @@ private enum State { this.cache = store.getMetadataCache(serde); this.serde = serde; this.executor = executor; + this.sequencer = FutureUtil.Sequencer.create(); store.registerSessionListener(this::handleSessionEvent); store.registerListener(this::handleDataNotification); } @@ -118,9 +120,8 @@ public CompletableFuture> acquireLock(String path, T value) { private void handleSessionEvent(SessionEvent se) { // We want to make sure we're processing one event at a time and that we're done with one event before going // for the next one. - executor.execute(SafeRunnable.safeRun(() -> { - List> futures = new ArrayList<>(); - + sequencer.sequential(() -> FutureUtil.composeAsync(() -> { + final List> futures = new ArrayList<>(); if (se == SessionEvent.SessionReestablished) { log.info("Metadata store session has been re-established. Revalidating all the existing locks."); for (ResourceLockImpl lock : locks.values()) { @@ -133,13 +134,12 @@ private void handleSessionEvent(SessionEvent se) { futures.add(lock.revalidateIfNeededAfterReconnection()); } } - - try { - FutureUtil.waitForAll(futures).get(); - } catch (ExecutionException | InterruptedException e) { - log.warn("Failure when processing session event", e); - } - })); + return FutureUtil.waitForAll(futures) + .exceptionally(ex -> { + log.warn("Failure when processing session event", ex); + return null; + }); + }, executor)); } private void handleDataNotification(Notification n) { From ee06266df19a345bab7d046db2aeedb6b2b98ed5 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Tue, 21 Mar 2023 10:39:12 +0800 Subject: [PATCH 2/3] Aovid create another future --- .../apache/pulsar/common/util/FutureUtil.java | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java index 8cde6e7c99127..2b082b4a7899b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java @@ -28,6 +28,7 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -239,12 +240,27 @@ public static CompletableFuture addTimeoutHandling(CompletableFuture f return future; } + /** + * @throws RejectedExecutionException if this task cannot be accepted for execution + * @throws NullPointerException if one of params is null + */ public static @Nonnull CompletableFuture composeAsync(Supplier> futureSupplier, Executor executor) { Objects.requireNonNull(futureSupplier); Objects.requireNonNull(executor); - return CompletableFuture.completedFuture(null) - .thenComposeAsync(__ -> futureSupplier.get(), executor); + final CompletableFuture future = new CompletableFuture<>(); + try { + executor.execute(() -> futureSupplier.get().whenComplete((result, error) -> { + if (error != null) { + future.completeExceptionally(error); + return; + } + future.complete(result); + })); + } catch (RejectedExecutionException ex) { + future.completeExceptionally(ex); + } + return future; } From ae1614a935ed7bf44cef68c030d0f3a2d9c5c536 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Tue, 21 Mar 2023 10:59:58 +0800 Subject: [PATCH 3/3] Fix checkstyle --- .../pulsar/metadata/coordination/impl/LeaderElectionImpl.java | 1 - .../pulsar/metadata/coordination/impl/LockManagerImpl.java | 2 -- 2 files changed, 3 deletions(-) diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java index b2ac4b6bb5bb9..11ae62226e7cd 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java @@ -24,7 +24,6 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java index 38ce58b2d3bc1..4da6b7998a0c4 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java @@ -27,11 +27,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; -import org.apache.bookkeeper.common.util.SafeRunnable; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.MetadataCache; import org.apache.pulsar.metadata.api.MetadataSerde;