From 542cc5a9b34a8f99433af5a8465915fb2c4e248b Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Wed, 8 Mar 2023 16:47:37 +0800 Subject: [PATCH 1/8] [fix][metadata] Fix notification thread block causes double owner. --- .../coordination/impl/LockManagerImpl.java | 40 +++++++--- .../coordination/impl/ResourceLockImpl.java | 73 ++++++++++++------- 2 files changed, 76 insertions(+), 37 deletions(-) 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 097f15af27677..58be07f619835 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 @@ -19,6 +19,7 @@ package org.apache.pulsar.metadata.coordination.impl; import com.fasterxml.jackson.databind.type.TypeFactory; +import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -29,6 +30,9 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.util.SafeRunnable; @@ -48,7 +52,7 @@ @Slf4j class LockManagerImpl implements LockManager { - + private static final Duration REVALIDATE_TIMEOUT = Duration.ofSeconds(30); private final Map> locks = new ConcurrentHashMap<>(); private final MetadataStoreExtended store; private final MetadataCache cache; @@ -118,28 +122,42 @@ 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<>(); - + final SafeRunnable task = SafeRunnable.safeRun(() -> { + 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()) { - futures.add(lock.revalidate(lock.getValue(), true)); + futures.add(lock.revalidateOnce(lock.getValue())); } - } else if (se == SessionEvent.Reconnected) { log.info("Metadata store connection has been re-established. Revalidating locks that were pending."); for (ResourceLockImpl lock : locks.values()) { futures.add(lock.revalidateIfNeededAfterReconnection()); } } - try { - FutureUtil.waitForAll(futures).get(); - } catch (ExecutionException | InterruptedException e) { - log.warn("Failure when processing session event", e); + FutureUtil.waitForAll(futures).get(REVALIDATE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + } catch (ExecutionException ex) { + log.warn("Got exception when execute revalidate ", ex); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + log.warn("Got thread interrupted exception when execute revalidate."); + } catch (TimeoutException ex) { + log.warn("Got timeout exception when execute revalidate"); + for (final CompletableFuture future : futures) { + if (!future.isDone()) { + if(!future.cancel(true)) { + log.warn("Failed to cancel the revalidation future {}", future); + } + } + } } - })); + }); + try { + executor.execute(task); + } catch (RejectedExecutionException ex) { + log.warn("Session events cannot be executed because the executor has been closed."); + } } private void handleDataNotification(Notification n) { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index dea9aa1acb90f..090becd2bb26b 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -20,6 +20,7 @@ import java.util.EnumSet; import java.util.Optional; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; @@ -32,6 +33,7 @@ import org.apache.pulsar.metadata.api.coordination.ResourceLock; import org.apache.pulsar.metadata.api.extended.CreateOption; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +import javax.annotation.Nonnull; @Slf4j public class ResourceLockImpl implements ResourceLock { @@ -128,7 +130,7 @@ synchronized CompletableFuture acquire(T newValue) { .thenRun(() -> result.complete(null)) .exceptionally(ex -> { if (ex.getCause() instanceof LockBusyException) { - revalidate(newValue, false) + revalidate(newValue) .thenAccept(__ -> result.complete(null)) .exceptionally(ex1 -> { result.completeExceptionally(ex1); @@ -185,21 +187,29 @@ synchronized void lockWasInvalidated() { } log.info("Lock on resource {} was invalidated", path); - revalidate(value, true) - .thenRun(() -> log.info("Successfully revalidated the lock on {}", path)); + revalidateOnce(value); } synchronized CompletableFuture revalidateIfNeededAfterReconnection() { if (revalidateAfterReconnection) { revalidateAfterReconnection = false; log.warn("Revalidate lock at {} after reconnection", path); - return revalidate(value, true); + return revalidateOnce(value); } else { return CompletableFuture.completedFuture(null); } } - synchronized CompletableFuture revalidate(T newValue, boolean revalidateAfterReconnection) { + /** + * Revalidate the distributed lock if it is not released. + * This method is thread-safe and it will perform multiple re-validation operations in turn. + * @param newValue the lock value + */ + synchronized @Nonnull CompletableFuture revalidate(@Nonnull T newValue) { + if (state == State.Released) { + // We don't need to revalidate the released lock since the expired future has been executed. + return CompletableFuture.completedFuture(null); + } if (revalidateFuture == null || revalidateFuture.isDone()) { revalidateFuture = doRevalidate(newValue); } else { @@ -217,30 +227,41 @@ synchronized CompletableFuture revalidate(T newValue, boolean revalidateAf }); revalidateFuture = newFuture; } - revalidateFuture.exceptionally(ex -> { - synchronized (ResourceLockImpl.this) { - Throwable realCause = FutureUtil.unwrapCompletionException(ex); - if (!revalidateAfterReconnection || realCause instanceof BadVersionException - || realCause instanceof LockBusyException) { - log.warn("Failed to revalidate the lock at {}. Marked as expired. {}", - path, realCause.getMessage()); - state = State.Released; - expiredFuture.complete(null); - } else { - // We failed to revalidate the lock due to connectivity issue - // Continue assuming we hold the lock, until we can revalidate it, either - // on Reconnected or SessionReestablished events. - ResourceLockImpl.this.revalidateAfterReconnection = true; - log.warn("Failed to revalidate the lock at {}. Retrying later on reconnection {}", path, - realCause.getMessage()); - } - } - return null; - }); return revalidateFuture; } - private synchronized CompletableFuture doRevalidate(T newValue) { + /** + * This method will auto mark the lock is released if revalidation operation got one of #{@code } + * @param newValue the lock value + */ + @Nonnull CompletableFuture revalidateOnce(@Nonnull T newValue) { + return revalidate(newValue) + .thenRun(() -> log.info("Successfully revalidated once the lock on {}", path)) + .exceptionally(ex -> { + synchronized (ResourceLockImpl.this) { + Throwable realCause = FutureUtil.unwrapCompletionException(ex); + if (realCause instanceof BadVersionException || realCause instanceof LockBusyException + // If the revalidation future is cancelled, + // we can assume the invoker will give up this lock in memory. + || realCause instanceof CancellationException) { + log.warn("Failed to revalidate the lock at {}. Marked as expired. {}", + path, realCause.getMessage()); + state = State.Released; + expiredFuture.complete(null); + } else { + // We failed to revalidate the lock due to connectivity issue + // Continue assuming we hold the lock, until we can revalidate it, either + // on Reconnected or SessionReestablished events. + ResourceLockImpl.this.revalidateAfterReconnection = true; + log.warn("Failed to revalidate the lock at {}. Retrying later on reconnection {}", path, + realCause.getMessage()); + } + } + return null; + }); + } + + private synchronized @Nonnull CompletableFuture doRevalidate(@Nonnull T newValue) { if (log.isDebugEnabled()) { log.debug("doRevalidate with newValue={}, version={}", newValue, version); } From 7adf86f2cb644b34012549700ccf4522747dbe22 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Wed, 8 Mar 2023 17:42:45 +0800 Subject: [PATCH 2/8] Fix the doc --- .../coordination/impl/LockManagerImpl.java | 2 +- .../coordination/impl/ResourceLockImpl.java | 15 +++++++++++---- 2 files changed, 12 insertions(+), 5 deletions(-) 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 58be07f619835..55eb3d0bb06f7 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 @@ -127,7 +127,7 @@ private void handleSessionEvent(SessionEvent se) { if (se == SessionEvent.SessionReestablished) { log.info("Metadata store session has been re-established. Revalidating all the existing locks."); for (ResourceLockImpl lock : locks.values()) { - futures.add(lock.revalidateOnce(lock.getValue())); + futures.add(lock.silentRevalidateOnce(lock.getValue())); } } else if (se == SessionEvent.Reconnected) { log.info("Metadata store connection has been re-established. Revalidating locks that were pending."); diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index 090becd2bb26b..15a2054bd6a12 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -187,14 +187,14 @@ synchronized void lockWasInvalidated() { } log.info("Lock on resource {} was invalidated", path); - revalidateOnce(value); + silentRevalidateOnce(value); } synchronized CompletableFuture revalidateIfNeededAfterReconnection() { if (revalidateAfterReconnection) { revalidateAfterReconnection = false; log.warn("Revalidate lock at {} after reconnection", path); - return revalidateOnce(value); + return silentRevalidateOnce(value); } else { return CompletableFuture.completedFuture(null); } @@ -231,10 +231,17 @@ synchronized CompletableFuture revalidateIfNeededAfterReconnection() { } /** - * This method will auto mark the lock is released if revalidation operation got one of #{@code } + * This method designed for background notification usage,it will auto mark the lock is released if revalidation + * operation got one of exceptions as follows: + * - LockBusyException + * - BadVersionException + * - CancellationException + * * @param newValue the lock value + * @return The revalidation future #Notice: It will not return any useful result, + * the caller needs to re-check the lock state after silent revalidation once. */ - @Nonnull CompletableFuture revalidateOnce(@Nonnull T newValue) { + @Nonnull CompletableFuture silentRevalidateOnce(@Nonnull T newValue) { return revalidate(newValue) .thenRun(() -> log.info("Successfully revalidated once the lock on {}", path)) .exceptionally(ex -> { From 1e4d63883dd6ef5078f87bda12a4eecfef602c67 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Wed, 8 Mar 2023 17:59:20 +0800 Subject: [PATCH 3/8] Fix checkstyle --- .../pulsar/metadata/coordination/impl/LockManagerImpl.java | 2 +- .../pulsar/metadata/coordination/impl/ResourceLockImpl.java | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) 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 55eb3d0bb06f7..a8e5af4f3a13d 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 @@ -146,7 +146,7 @@ private void handleSessionEvent(SessionEvent se) { log.warn("Got timeout exception when execute revalidate"); for (final CompletableFuture future : futures) { if (!future.isDone()) { - if(!future.cancel(true)) { + if (!future.cancel(true)) { log.warn("Failed to cancel the revalidation future {}", future); } } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index 15a2054bd6a12..ecaa91e9f6078 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -22,6 +22,7 @@ import java.util.Optional; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; +import javax.annotation.Nonnull; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.pulsar.common.util.FutureUtil; @@ -33,7 +34,6 @@ import org.apache.pulsar.metadata.api.coordination.ResourceLock; import org.apache.pulsar.metadata.api.extended.CreateOption; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; -import javax.annotation.Nonnull; @Slf4j public class ResourceLockImpl implements ResourceLock { @@ -231,8 +231,8 @@ synchronized CompletableFuture revalidateIfNeededAfterReconnection() { } /** - * This method designed for background notification usage,it will auto mark the lock is released if revalidation - * operation got one of exceptions as follows: + * This method designed for background notification usage. + * It will auto mark the lock is released if revalidation operation got one of exceptions as follows: * - LockBusyException * - BadVersionException * - CancellationException From a098e17cd92ec88f0f4614fcf2b40c556c2effb6 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Thu, 9 Mar 2023 00:53:10 +0800 Subject: [PATCH 4/8] [fix][meta] Fix deadlock causes session notification is not to work --- .../coordination/impl/LeaderElectionImpl.java | 10 ++- .../impl/LeaderElectionImplTest.java | 65 +++++++++++++++++++ 2 files changed, 73 insertions(+), 2 deletions(-) create mode 100644 pulsar-metadata/src/test/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImplTest.java 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 409d49dcd26ab..0f0f3399d4f19 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 @@ -28,6 +28,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import com.google.common.annotations.VisibleForTesting; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.common.util.SafeRunnable; @@ -111,13 +112,13 @@ private synchronized CompletableFuture elect() { } else { return tryToBecomeLeader(); } - }).thenComposeAsync(leaderElectionState -> { + }).thenCompose(leaderElectionState -> { // make sure that the cache contains the current leader // so that getLeaderValueIfPresent works on all brokers cache.refresh(path); return cache.get(path) .thenApply(__ -> leaderElectionState); - }, executor); + }); } private synchronized CompletableFuture handleExistingLeaderValue(GetResult res) { @@ -336,4 +337,9 @@ private void handlePathNotification(Notification notification) { } } } + + @VisibleForTesting + protected ScheduledExecutorService getSchedulerExecutor() { + return executor; + } } diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImplTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImplTest.java new file mode 100644 index 0000000000000..027521d2ffc17 --- /dev/null +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImplTest.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.metadata.coordination.impl; + +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; +import lombok.Cleanup; +import org.apache.pulsar.metadata.BaseMetadataStoreTest; +import org.apache.pulsar.metadata.api.MetadataStoreConfig; +import org.apache.pulsar.metadata.api.coordination.CoordinationService; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +import org.testng.annotations.Test; + +public class LeaderElectionImplTest extends BaseMetadataStoreTest { + + @Test(dataProvider = "impl", timeOut = 10000) + public void validateDeadLock(String provider, Supplier urlSupplier) + throws Exception { + if (provider.equals("Memory") || provider.equals("RocksDB")) { + // There are no multiple sessions for the local memory provider + return; + } + + @Cleanup + MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(), + MetadataStoreConfig.builder().build()); + + String path = newKey(); + + @Cleanup + CoordinationService cs = new CoordinationServiceImpl(store); + + @Cleanup + LeaderElectionImpl le = (LeaderElectionImpl) cs.getLeaderElection(String.class, + path, __ -> { + }); + final CompletableFuture blockFuture = new CompletableFuture<>(); + // simulate handleSessionNotification method logic + le.getSchedulerExecutor().execute(() -> { + try { + le.elect("test-2").join(); + blockFuture.complete(null); + } catch (Throwable ex) { + blockFuture.completeExceptionally(ex); + } + }); + blockFuture.join(); + } +} From 0c53166d84cf5cb7b9c16297236f0f48cbfd0b87 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Thu, 9 Mar 2023 00:54:59 +0800 Subject: [PATCH 5/8] Revert "Fix checkstyle" This reverts commit 1e4d63883dd6ef5078f87bda12a4eecfef602c67. --- .../pulsar/metadata/coordination/impl/LockManagerImpl.java | 2 +- .../pulsar/metadata/coordination/impl/ResourceLockImpl.java | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) 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 a8e5af4f3a13d..55eb3d0bb06f7 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 @@ -146,7 +146,7 @@ private void handleSessionEvent(SessionEvent se) { log.warn("Got timeout exception when execute revalidate"); for (final CompletableFuture future : futures) { if (!future.isDone()) { - if (!future.cancel(true)) { + if(!future.cancel(true)) { log.warn("Failed to cancel the revalidation future {}", future); } } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index ecaa91e9f6078..15a2054bd6a12 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -22,7 +22,6 @@ import java.util.Optional; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; -import javax.annotation.Nonnull; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.pulsar.common.util.FutureUtil; @@ -34,6 +33,7 @@ import org.apache.pulsar.metadata.api.coordination.ResourceLock; import org.apache.pulsar.metadata.api.extended.CreateOption; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +import javax.annotation.Nonnull; @Slf4j public class ResourceLockImpl implements ResourceLock { @@ -231,8 +231,8 @@ synchronized CompletableFuture revalidateIfNeededAfterReconnection() { } /** - * This method designed for background notification usage. - * It will auto mark the lock is released if revalidation operation got one of exceptions as follows: + * This method designed for background notification usage,it will auto mark the lock is released if revalidation + * operation got one of exceptions as follows: * - LockBusyException * - BadVersionException * - CancellationException From a3950035344056b2ed507a2b255c781e52472b46 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Thu, 9 Mar 2023 00:54:59 +0800 Subject: [PATCH 6/8] Revert "Fix the doc" This reverts commit 7adf86f2cb644b34012549700ccf4522747dbe22. --- .../coordination/impl/LockManagerImpl.java | 2 +- .../coordination/impl/ResourceLockImpl.java | 15 ++++----------- 2 files changed, 5 insertions(+), 12 deletions(-) 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 55eb3d0bb06f7..58be07f619835 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 @@ -127,7 +127,7 @@ private void handleSessionEvent(SessionEvent se) { if (se == SessionEvent.SessionReestablished) { log.info("Metadata store session has been re-established. Revalidating all the existing locks."); for (ResourceLockImpl lock : locks.values()) { - futures.add(lock.silentRevalidateOnce(lock.getValue())); + futures.add(lock.revalidateOnce(lock.getValue())); } } else if (se == SessionEvent.Reconnected) { log.info("Metadata store connection has been re-established. Revalidating locks that were pending."); diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index 15a2054bd6a12..090becd2bb26b 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -187,14 +187,14 @@ synchronized void lockWasInvalidated() { } log.info("Lock on resource {} was invalidated", path); - silentRevalidateOnce(value); + revalidateOnce(value); } synchronized CompletableFuture revalidateIfNeededAfterReconnection() { if (revalidateAfterReconnection) { revalidateAfterReconnection = false; log.warn("Revalidate lock at {} after reconnection", path); - return silentRevalidateOnce(value); + return revalidateOnce(value); } else { return CompletableFuture.completedFuture(null); } @@ -231,17 +231,10 @@ synchronized CompletableFuture revalidateIfNeededAfterReconnection() { } /** - * This method designed for background notification usage,it will auto mark the lock is released if revalidation - * operation got one of exceptions as follows: - * - LockBusyException - * - BadVersionException - * - CancellationException - * + * This method will auto mark the lock is released if revalidation operation got one of #{@code } * @param newValue the lock value - * @return The revalidation future #Notice: It will not return any useful result, - * the caller needs to re-check the lock state after silent revalidation once. */ - @Nonnull CompletableFuture silentRevalidateOnce(@Nonnull T newValue) { + @Nonnull CompletableFuture revalidateOnce(@Nonnull T newValue) { return revalidate(newValue) .thenRun(() -> log.info("Successfully revalidated once the lock on {}", path)) .exceptionally(ex -> { From a377e6f46adb1d6907db7f05c5a9f2f720e69c63 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Thu, 9 Mar 2023 00:54:59 +0800 Subject: [PATCH 7/8] Revert "[fix][metadata] Fix notification thread block causes double owner." This reverts commit 542cc5a9b34a8f99433af5a8465915fb2c4e248b. --- .../coordination/impl/LockManagerImpl.java | 40 +++------- .../coordination/impl/ResourceLockImpl.java | 73 +++++++------------ 2 files changed, 37 insertions(+), 76 deletions(-) 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 58be07f619835..097f15af27677 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 @@ -19,7 +19,6 @@ package org.apache.pulsar.metadata.coordination.impl; import com.fasterxml.jackson.databind.type.TypeFactory; -import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -30,9 +29,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.util.SafeRunnable; @@ -52,7 +48,7 @@ @Slf4j class LockManagerImpl implements LockManager { - private static final Duration REVALIDATE_TIMEOUT = Duration.ofSeconds(30); + private final Map> locks = new ConcurrentHashMap<>(); private final MetadataStoreExtended store; private final MetadataCache cache; @@ -122,42 +118,28 @@ 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. - final SafeRunnable task = SafeRunnable.safeRun(() -> { - final List> futures = new ArrayList<>(); + executor.execute(SafeRunnable.safeRun(() -> { + 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()) { - futures.add(lock.revalidateOnce(lock.getValue())); + futures.add(lock.revalidate(lock.getValue(), true)); } + } else if (se == SessionEvent.Reconnected) { log.info("Metadata store connection has been re-established. Revalidating locks that were pending."); for (ResourceLockImpl lock : locks.values()) { futures.add(lock.revalidateIfNeededAfterReconnection()); } } + try { - FutureUtil.waitForAll(futures).get(REVALIDATE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); - } catch (ExecutionException ex) { - log.warn("Got exception when execute revalidate ", ex); - } catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - log.warn("Got thread interrupted exception when execute revalidate."); - } catch (TimeoutException ex) { - log.warn("Got timeout exception when execute revalidate"); - for (final CompletableFuture future : futures) { - if (!future.isDone()) { - if(!future.cancel(true)) { - log.warn("Failed to cancel the revalidation future {}", future); - } - } - } + FutureUtil.waitForAll(futures).get(); + } catch (ExecutionException | InterruptedException e) { + log.warn("Failure when processing session event", e); } - }); - try { - executor.execute(task); - } catch (RejectedExecutionException ex) { - log.warn("Session events cannot be executed because the executor has been closed."); - } + })); } private void handleDataNotification(Notification n) { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index 090becd2bb26b..dea9aa1acb90f 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -20,7 +20,6 @@ import java.util.EnumSet; import java.util.Optional; -import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; @@ -33,7 +32,6 @@ import org.apache.pulsar.metadata.api.coordination.ResourceLock; import org.apache.pulsar.metadata.api.extended.CreateOption; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; -import javax.annotation.Nonnull; @Slf4j public class ResourceLockImpl implements ResourceLock { @@ -130,7 +128,7 @@ synchronized CompletableFuture acquire(T newValue) { .thenRun(() -> result.complete(null)) .exceptionally(ex -> { if (ex.getCause() instanceof LockBusyException) { - revalidate(newValue) + revalidate(newValue, false) .thenAccept(__ -> result.complete(null)) .exceptionally(ex1 -> { result.completeExceptionally(ex1); @@ -187,29 +185,21 @@ synchronized void lockWasInvalidated() { } log.info("Lock on resource {} was invalidated", path); - revalidateOnce(value); + revalidate(value, true) + .thenRun(() -> log.info("Successfully revalidated the lock on {}", path)); } synchronized CompletableFuture revalidateIfNeededAfterReconnection() { if (revalidateAfterReconnection) { revalidateAfterReconnection = false; log.warn("Revalidate lock at {} after reconnection", path); - return revalidateOnce(value); + return revalidate(value, true); } else { return CompletableFuture.completedFuture(null); } } - /** - * Revalidate the distributed lock if it is not released. - * This method is thread-safe and it will perform multiple re-validation operations in turn. - * @param newValue the lock value - */ - synchronized @Nonnull CompletableFuture revalidate(@Nonnull T newValue) { - if (state == State.Released) { - // We don't need to revalidate the released lock since the expired future has been executed. - return CompletableFuture.completedFuture(null); - } + synchronized CompletableFuture revalidate(T newValue, boolean revalidateAfterReconnection) { if (revalidateFuture == null || revalidateFuture.isDone()) { revalidateFuture = doRevalidate(newValue); } else { @@ -227,41 +217,30 @@ synchronized CompletableFuture revalidateIfNeededAfterReconnection() { }); revalidateFuture = newFuture; } + revalidateFuture.exceptionally(ex -> { + synchronized (ResourceLockImpl.this) { + Throwable realCause = FutureUtil.unwrapCompletionException(ex); + if (!revalidateAfterReconnection || realCause instanceof BadVersionException + || realCause instanceof LockBusyException) { + log.warn("Failed to revalidate the lock at {}. Marked as expired. {}", + path, realCause.getMessage()); + state = State.Released; + expiredFuture.complete(null); + } else { + // We failed to revalidate the lock due to connectivity issue + // Continue assuming we hold the lock, until we can revalidate it, either + // on Reconnected or SessionReestablished events. + ResourceLockImpl.this.revalidateAfterReconnection = true; + log.warn("Failed to revalidate the lock at {}. Retrying later on reconnection {}", path, + realCause.getMessage()); + } + } + return null; + }); return revalidateFuture; } - /** - * This method will auto mark the lock is released if revalidation operation got one of #{@code } - * @param newValue the lock value - */ - @Nonnull CompletableFuture revalidateOnce(@Nonnull T newValue) { - return revalidate(newValue) - .thenRun(() -> log.info("Successfully revalidated once the lock on {}", path)) - .exceptionally(ex -> { - synchronized (ResourceLockImpl.this) { - Throwable realCause = FutureUtil.unwrapCompletionException(ex); - if (realCause instanceof BadVersionException || realCause instanceof LockBusyException - // If the revalidation future is cancelled, - // we can assume the invoker will give up this lock in memory. - || realCause instanceof CancellationException) { - log.warn("Failed to revalidate the lock at {}. Marked as expired. {}", - path, realCause.getMessage()); - state = State.Released; - expiredFuture.complete(null); - } else { - // We failed to revalidate the lock due to connectivity issue - // Continue assuming we hold the lock, until we can revalidate it, either - // on Reconnected or SessionReestablished events. - ResourceLockImpl.this.revalidateAfterReconnection = true; - log.warn("Failed to revalidate the lock at {}. Retrying later on reconnection {}", path, - realCause.getMessage()); - } - } - return null; - }); - } - - private synchronized @Nonnull CompletableFuture doRevalidate(@Nonnull T newValue) { + private synchronized CompletableFuture doRevalidate(T newValue) { if (log.isDebugEnabled()) { log.debug("doRevalidate with newValue={}, version={}", newValue, version); } From 80d061312ea1dbf0f08abd5931d45869dbc0ee8d Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Thu, 9 Mar 2023 01:06:04 +0800 Subject: [PATCH 8/8] Fix checkstyle --- .../pulsar/metadata/coordination/impl/LeaderElectionImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 0f0f3399d4f19..ad2a5bef70610 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 @@ -19,6 +19,7 @@ package org.apache.pulsar.metadata.coordination.impl; import com.fasterxml.jackson.databind.type.TypeFactory; +import com.google.common.annotations.VisibleForTesting; import java.util.EnumSet; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -28,7 +29,6 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; -import com.google.common.annotations.VisibleForTesting; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.common.util.SafeRunnable;