Skip to content

Issue #2854: Use correct lockpath, store Lock to heldLocks at ZkLedgerUnderreplicationManager#acquireUnderreplicatedLedger #2855

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

Merged
merlimat merged 1 commit into from
Nov 2, 2021
Merged

Issue #2854: Use correct lockpath, store Lock to heldLocks at ZkLedgerUnderreplicationManager#acquireUnderreplicatedLedger #2855

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
57 changes: 32 additions & 25 deletions ...eper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -94,9 +95,9 @@ public class ZkLedgerUnderreplicationManager implements LedgerUnderreplicationMa

private static class Lock {
private final String lockZNode;
private final int ledgerZNodeVersion;
private final Optional<Integer> ledgerZNodeVersion;

Lock(String lockZNode, int ledgerZNodeVersion) {
Lock(String lockZNode, Optional<Integer> ledgerZNodeVersion) {
this.lockZNode = lockZNode;
this.ledgerZNodeVersion = ledgerZNodeVersion;
}
Expand All @@ -105,13 +106,14 @@ String getLockZNode() {
return lockZNode;
}

int getLedgerZNodeVersion() {
Optional<Integer> getLedgerZNodeVersion() {
return ledgerZNodeVersion;
}
}
private final Map<Long, Lock> heldLocks = new ConcurrentHashMap<Long, Lock>();
private final Pattern idExtractionPattern;

private final String rootPath;
private final String basePath;
private final String urLedgerPath;
private final String urLockPath;
Expand All @@ -127,7 +129,8 @@ int getLedgerZNodeVersion() {
public ZkLedgerUnderreplicationManager(AbstractConfiguration conf, ZooKeeper zkc)
throws KeeperException, InterruptedException, ReplicationException.CompatibilityException {
this.conf = conf;
basePath = getBasePath(ZKMetadataDriverBase.resolveZkLedgersRootPath(conf));
rootPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf);
basePath = getBasePath(rootPath);
layoutZNode = basePath + '/' + BookKeeperConstants.LAYOUT_ZNODE;
urLedgerPath = basePath
+ BookKeeperConstants.DEFAULT_ZK_LEDGERS_ROOT_PATH;
Expand Down Expand Up @@ -389,24 +392,27 @@ public void markLedgerReplicated(long ledgerId) throws ReplicationException.Unav
try {
Lock l = heldLocks.get(ledgerId);
if (l != null) {
zkc.delete(getUrLedgerZnode(ledgerId), l.getLedgerZNodeVersion());
final Optional<Integer> ledgerZNodeVersion = l.getLedgerZNodeVersion();
if (ledgerZNodeVersion.isPresent()) {
zkc.delete(getUrLedgerZnode(ledgerId), ledgerZNodeVersion.get());

try {
// clean up the hierarchy
String[] parts = getUrLedgerZnode(ledgerId).split("/");
for (int i = 1; i <= 4; i++) {
String[] p = Arrays.copyOf(parts, parts.length - i);
String path = Joiner.on("/").join(p);
Stat s = zkc.exists(path, null);
if (s != null) {
zkc.delete(path, s.getVersion());
try {
// clean up the hierarchy
String[] parts = getUrLedgerZnode(ledgerId).split("/");
for (int i = 1; i <= 4; i++) {
String[] p = Arrays.copyOf(parts, parts.length - i);
String path = Joiner.on("/").join(p);
Stat s = zkc.exists(path, null);
if (s != null) {
zkc.delete(path, s.getVersion());
}
}
} catch (KeeperException.NotEmptyException nee) {
// This can happen when cleaning up the hierarchy.
// It's safe to ignore, it simply means another
// ledger in the same hierarchy has been marked as
// underreplicated.
}
} catch (KeeperException.NotEmptyException nee) {
// This can happen when cleaning up the hierarchy.
// It's safe to ignore, it simply means another
// ledger in the same hierarchy has been marked as
// underreplicated.
}
}
} catch (KeeperException.NoNodeException nne) {
Expand Down Expand Up @@ -528,7 +534,7 @@ private long getLedgerToRereplicateFromHierarchy(String parent, long depth)
String lockPath = urLockPath + "/" + tryChild;
long ledgerId = getLedgerId(tryChild);
zkc.create(lockPath, LOCK_DATA, zkAcls, CreateMode.EPHEMERAL);
heldLocks.put(ledgerId, new Lock(lockPath, stat.getVersion()));
heldLocks.put(ledgerId, new Lock(lockPath, Optional.of(stat.getVersion())));
return ledgerId;
} catch (KeeperException.NodeExistsException nee) {
children.remove(tryChild);
Expand Down Expand Up @@ -782,19 +788,20 @@ public boolean isLedgerBeingReplicated(long ledgerId) throws ReplicationExceptio
/**
* Acquire the underreplicated ledger lock.
*/
public static void acquireUnderreplicatedLedgerLock(ZooKeeper zkc, String zkLedgersRootPath,
public static String acquireUnderreplicatedLedgerLock(ZooKeeper zkc, String zkLedgersRootPath,
long ledgerId, List<ACL> zkAcls)
throws KeeperException, InterruptedException {
ZkUtils.createFullPathOptimistic(zkc, getUrLedgerLockZnode(getUrLockPath(zkLedgersRootPath), ledgerId),
LOCK_DATA, zkAcls, CreateMode.EPHEMERAL);
final String lockPath = getUrLedgerLockZnode(getUrLockPath(zkLedgersRootPath), ledgerId);
ZkUtils.createFullPathOptimistic(zkc, lockPath, LOCK_DATA, zkAcls, CreateMode.EPHEMERAL);
return lockPath;
}

@Override
public void acquireUnderreplicatedLedger(long ledgerId)
throws ReplicationException {
try {
acquireUnderreplicatedLedgerLock(zkc, getUrLedgerLockZnode(urLockPath, ledgerId), ledgerId,
ZkUtils.getACLs(conf));
final String lockPath = acquireUnderreplicatedLedgerLock(zkc, rootPath, ledgerId, ZkUtils.getACLs(conf));
heldLocks.put(ledgerId, new Lock(lockPath, Optional.empty()));
} catch (Exception e) {
throw new ReplicationException.UnavailableException(
"Failed to acquire underreplicated ledger lock for " + ledgerId, e);
Expand Down
32 changes: 32 additions & 0 deletions bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GcOverreplicatedLedgerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,13 @@

import com.google.common.collect.Lists;
import java.io.IOException;
import java.net.URI;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.SortedMap;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.bookkeeper.bookie.GarbageCollector.GarbageCleaner;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerHandle;
Expand All @@ -36,11 +38,16 @@
import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerManagerTestCase;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.meta.MetadataDrivers;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.meta.exceptions.MetadataException;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.SnapshotMap;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.zookeeper.ZooDefs;
import org.junit.Assert;
import org.junit.Before;
Expand Down Expand Up @@ -81,6 +88,16 @@ public void testGcOverreplicatedLedger() throws Exception {

BookieId bookieNotInEnsemble = getBookieNotInEnsemble(newLedgerMetadata);
ServerConfiguration bkConf = getBkConf(bookieNotInEnsemble);

@Cleanup
final MetadataBookieDriver metadataDriver = instantiateMetadataDriver(bkConf);
@Cleanup
final LedgerManagerFactory lmf = metadataDriver.getLedgerManagerFactory();
@Cleanup
final LedgerUnderreplicationManager lum = lmf.newLedgerUnderreplicationManager();

Assert.assertFalse(lum.isLedgerBeingReplicated(lh.getId()));

bkConf.setGcOverreplicatedLedgerWaitTime(10, TimeUnit.MILLISECONDS);

lh.close();
Expand All @@ -102,9 +119,24 @@ public void clean(long ledgerId) {
}
});

Assert.assertFalse(lum.isLedgerBeingReplicated(lh.getId()));
Assert.assertFalse(activeLedgers.containsKey(lh.getId()));
}

private static MetadataBookieDriver instantiateMetadataDriver(ServerConfiguration conf)
throws BookieException {
try {
final String metadataServiceUriStr = conf.getMetadataServiceUri();
final MetadataBookieDriver driver = MetadataDrivers.getBookieDriver(URI.create(metadataServiceUriStr));
driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
return driver;
} catch (MetadataException me) {
throw new BookieException.MetadataStoreException("Failed to initialize metadata bookie driver", me);
} catch (ConfigurationException e) {
throw new BookieException.BookieIllegalOpException(e);
}
}

@Test
public void testNoGcOfLedger() throws Exception {
LedgerHandle lh = bkc.createLedger(2, 2, DigestType.MAC, "".getBytes());
Expand Down