diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java index 680264af6d6..b340e0721cb 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java @@ -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; @@ -94,9 +95,9 @@ public class ZkLedgerUnderreplicationManager implements LedgerUnderreplicationMa private static class Lock { private final String lockZNode; - private final int ledgerZNodeVersion; + private final Optional ledgerZNodeVersion; - Lock(String lockZNode, int ledgerZNodeVersion) { + Lock(String lockZNode, Optional ledgerZNodeVersion) { this.lockZNode = lockZNode; this.ledgerZNodeVersion = ledgerZNodeVersion; } @@ -105,13 +106,14 @@ String getLockZNode() { return lockZNode; } - int getLedgerZNodeVersion() { + Optional getLedgerZNodeVersion() { return ledgerZNodeVersion; } } private final Map heldLocks = new ConcurrentHashMap(); private final Pattern idExtractionPattern; + private final String rootPath; private final String basePath; private final String urLedgerPath; private final String urLockPath; @@ -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; @@ -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 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) { @@ -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); @@ -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 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); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GcOverreplicatedLedgerTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GcOverreplicatedLedgerTest.java index 35dee8756c9..afe5ec899cf 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GcOverreplicatedLedgerTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GcOverreplicatedLedgerTest.java @@ -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; @@ -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; @@ -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(); @@ -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());