Skip to content

[branch-2.8][cherry-pick] Cleanup state when lock revalidation gets LockBusyException #19135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lordcheng10 wants to merge 4 commits into from
Closed

[branch-2.8][cherry-pick] Cleanup state when lock revalidation gets LockBusyException #19135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
27ddd22
chery pick 17700 to 2.8 branch
Jan 5, 2023
519857a
add test testCleanUpStateWhenRevalidationGotLockBusy
Jan 5, 2023
a19e861
update testCleanUpStateWhenRevalidationGotLockBusy
Jan 5, 2023
5c583ec
update test testCleanUpStateWhenRevalidationGotLockBusy
Jan 5, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion ...-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,9 @@ public CompletableFuture<ResourceLock<T>> acquireLock(String path, T value) {
private void handleSessionEvent(SessionEvent se) {
if (se == SessionEvent.SessionReestablished) {
log.info("Metadata store session has been re-established. Revalidating all the existing locks.");
locks.values().forEach(ResourceLockImpl::revalidate);
for (ResourceLockImpl<T> lock : locks.values()) {
lock.revalidate(true);
}
} else if (se == SessionEvent.Reconnected) {
log.info("Metadata store connection has been re-established. Revalidating locks that were pending.");
locks.values().forEach(ResourceLockImpl::revalidateIfNeededAfterReconnection);
Expand Down
53 changes: 32 additions & 21 deletions ...metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.GetResult;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
Expand Down Expand Up @@ -141,7 +142,7 @@ synchronized CompletableFuture<Void> acquire() {
.thenRun(() -> result.complete(null))
.exceptionally(ex -> {
if (ex.getCause() instanceof LockBusyException) {
revalidate()
revalidate(false)
.thenAccept(__ -> result.complete(null))
.exceptionally(ex1 -> {
result.completeExceptionally(ex1);
Expand Down Expand Up @@ -194,35 +195,19 @@ synchronized void lockWasInvalidated() {
}

log.info("Lock on resource {} was invalidated", path);
revalidate()
.thenRun(() -> log.info("Successfully revalidated the lock on {}", path))
.exceptionally(ex -> {
synchronized (ResourceLockImpl.this) {
if (ex.getCause() instanceof BadVersionException) {
log.warn("Failed to revalidate the lock at {}. Marked as expired", path);
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.
log.warn("Failed to revalidate the lock at {}. Retrying later on reconnection", path,
ex.getCause().getMessage());
}
}
return null;
});
revalidate(true)
.thenRun(() -> log.info("Successfully revalidated the lock on {}", path));
}

synchronized void revalidateIfNeededAfterReconnection() {
if (revalidateAfterReconnection) {
revalidateAfterReconnection = false;
log.warn("Revalidate lock at {} after reconnection", path);
revalidate();
revalidate(true);
}
}

synchronized CompletableFuture<Void> revalidate() {
private synchronized CompletableFuture<Void> doRevalidate() {
return store.get(path)
.thenCompose(optGetResult -> {
if (!optGetResult.isPresent()) {
Expand Down Expand Up @@ -279,4 +264,30 @@ synchronized CompletableFuture<Void> revalidate() {
}
});
}

synchronized CompletableFuture<Void> revalidate(boolean revalidateAfterReconnection) {
CompletableFuture<Void> revalidateFuture = doRevalidate();
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;
}
}
63 changes: 59 additions & 4 deletions pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,6 @@
*/
package org.apache.pulsar.metadata;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.fail;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
Expand All @@ -31,6 +27,8 @@
import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import lombok.Cleanup;

Expand All @@ -45,8 +43,12 @@
import org.apache.pulsar.metadata.api.extended.CreateOption;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
import org.awaitility.Awaitility;
import org.testng.annotations.Test;

import static org.testng.Assert.*;
import static org.testng.Assert.assertTrue;

public class LockManagerTest extends BaseMetadataStoreTest {

@Test(dataProvider = "impl")
Expand Down Expand Up @@ -228,4 +230,57 @@ public void revalidateLockOnDifferentSession(String provider, String url) throws

assertEquals(new String(store1.get(path2).join().get().getValue()), "\"value-1\"");
}

@Test(dataProvider = "impl")
public void testCleanUpStateWhenRevalidationGotLockBusy(String provider, String url)
throws Exception {

if (provider.equals("Memory") || provider.equals("RocksDB")) {
// Local memory provider doesn't really have the concept of multiple sessions
return;
}

@Cleanup
MetadataStoreExtended store = MetadataStoreExtended.create(url,
MetadataStoreConfig.builder().build());

@Cleanup
CoordinationService cs1 = new CoordinationServiceImpl(store);
@Cleanup
LockManager<String> lm1 = cs1.getLockManager(String.class);

@Cleanup
CoordinationService cs2 = new CoordinationServiceImpl(store);
@Cleanup
LockManager<String> lm2 = cs2.getLockManager(String.class);

String path1 = newKey();

ResourceLock<String> lock1 = lm1.acquireLock(path1, "value-1").join();
AtomicReference<ResourceLock<String>> lock2 = new AtomicReference<>();
// lock 2 will steal the distributed lock first.
Awaitility.await().until(()-> {
// Ensure steal the lock success.
try {
lock2.set(lm2.acquireLock(path1, "value-1").join());
return true;
} catch (Exception ex) {
return false;
}
});

// Since we can steal the lock repeatedly, we don't know which one will get it.
// But we can verify the final state.
Awaitility.await().untilAsserted(() -> {
if (lock1.getLockExpiredFuture().isDone()) {
assertTrue(lm1.listLocks(path1).join().isEmpty());
assertFalse(lock2.get().getLockExpiredFuture().isDone());
} else if (lock2.get().getLockExpiredFuture().isDone()) {
assertTrue(lm2.listLocks(path1).join().isEmpty());
assertFalse(lock1.getLockExpiredFuture().isDone());
} else {
fail("unexpected behaviour");
}
});
}
}