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..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; @@ -36,6 +37,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 +240,30 @@ 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); + 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; + } + + /** * 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..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; @@ -32,6 +31,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 +62,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 +86,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 +278,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..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; @@ -53,6 +51,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 +71,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 +118,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 +132,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) {