Repository: bookkeeper Updated Branches: refs/heads/master 13c5c88f6 -> 07852d358
BOOKKEEPER-1096: recursive znode delete When ledger is deleted, along with leaf node all the eligible branch nodes should be deleted in ZooKeeper. Author: Charan Reddy Guttapalem <[email protected]> Reviewers: Sijie Guo <[email protected]> Closes #186 from reddycharan/recursiveznodedelete Project: http://git-wip-us.apache.org/repos/asf/bookkeeper/repo Commit: http://git-wip-us.apache.org/repos/asf/bookkeeper/commit/07852d35 Tree: http://git-wip-us.apache.org/repos/asf/bookkeeper/tree/07852d35 Diff: http://git-wip-us.apache.org/repos/asf/bookkeeper/diff/07852d35 Branch: refs/heads/master Commit: 07852d35856dca232450135913090bac27b29abe Parents: 13c5c88 Author: Charan Reddy Guttapalem <[email protected]> Authored: Mon Jun 12 08:20:53 2017 -0700 Committer: Sijie Guo <[email protected]> Committed: Mon Jun 12 08:20:53 2017 -0700 ---------------------------------------------------------------------- .../meta/AbstractZkLedgerManager.java | 31 +++- .../org/apache/bookkeeper/util/ZkUtils.java | 68 +++++++- .../bookkeeper/meta/LedgerManagerTestCase.java | 1 + .../meta/LedgerMetadataCreationTest.java | 172 +++++++++++++++++++ .../org/apache/bookkeeper/util/TestZkUtils.java | 87 ++++++++++ 5 files changed, 346 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/07852d35/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java index 54fc78b..f3fd9e0 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java @@ -259,11 +259,15 @@ abstract class AbstractZkLedgerManager implements LedgerManager, Watcher { } /** - * Removes ledger metadata from ZooKeeper if version matches. + * Removes ledger metadata from ZooKeeper and deletes its parent znodes + * recursively if they dont have anymore children. * - * @param ledgerId ledger identifier - * @param version local version of metadata znode - * @param cb callback object + * @param ledgerId + * ledger identifier + * @param version + * local version of metadata znode + * @param cb + * callback object */ @Override public void removeLedgerMetadata(final long ledgerId, final Version version, @@ -282,8 +286,8 @@ abstract class AbstractZkLedgerManager implements LedgerManager, Watcher { znodeVersion = ((ZkVersion)version).getZnodeVersion(); } } - - zk.delete(getLedgerPath(ledgerId), znodeVersion, new VoidCallback() { + + VoidCallback callbackForDelete = new VoidCallback() { @Override public void processResult(int rc, String path, Object ctx) { int bkRc; @@ -308,9 +312,20 @@ abstract class AbstractZkLedgerManager implements LedgerManager, Watcher { } else { bkRc = BKException.Code.ZKException; } - cb.operationComplete(bkRc, (Void)null); + cb.operationComplete(bkRc, (Void) null); } - }, null); + }; + String ledgerZnodePath = getLedgerPath(ledgerId); + if (this instanceof HierarchicalLedgerManager || this instanceof LongHierarchicalLedgerManager) { + /* + * do recursive deletes only for HierarchicalLedgerManager and + * LongHierarchicalLedgerManager + */ + ZkUtils.asyncDeleteFullPathOptimistic(zk, ledgerZnodePath, znodeVersion, callbackForDelete, + ledgerZnodePath); + } else { + zk.delete(ledgerZnodePath, znodeVersion, callbackForDelete, null); + } } @Override http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/07852d35/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java index 24c0a52..a12ba40 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java @@ -29,17 +29,16 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.bookkeeper.conf.AbstractConfiguration; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback; -import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy; -import org.apache.bookkeeper.zookeeper.ZooKeeperClient; -import org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase; -import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.AsyncCallback; -import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.AsyncCallback.StringCallback; +import org.apache.zookeeper.AsyncCallback.VoidCallback; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.ACL; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -107,6 +106,48 @@ public class ZkUtils { } /** + * Asynchronously deletes zookeeper path recursively and optimistically. + * This method is used for deleting the leaf nodes and its corresponding + * parents if they don't have anymore children after deleting the child + * node. For this to work as expected, provided znodeVersion should be -1, + * so that there wont be version mismatches with any of the parent nodes. If + * it fails to delete the leafnode then it will callback with the received + * error code, but it fails to delete the parent node for whatsoever reason + * it stops proceeding further and it will callback with ok error code. + * + * @param zk + * Zookeeper client + * @param originalPath + * Zookeeper full path + * @param znodeVersion + * version of the node + * @param callback + * callback + * @param leafNodePath + * for actual caller this leafNodePath should be same as the + * originalPath. But when it is called recursively leafNodePath + * remains the same, but the originalPath will be internal nodes. + */ + public static void asyncDeleteFullPathOptimistic(final ZooKeeper zk, final String originalPath, + int znodeVersion, final AsyncCallback.VoidCallback callback, final String leafNodePath) { + zk.delete(originalPath, znodeVersion, new VoidCallback() { + @Override + public void processResult(int rc, String path, Object ctx) { + if (rc == Code.OK.intValue()) { + String parent = new File(originalPath).getParent().replace("\\", "/"); + asyncDeleteFullPathOptimistic(zk, parent, znodeVersion, callback, leafNodePath); + } else { + if (path.equals(leafNodePath)) { + callback.processResult(rc, path, leafNodePath); + } else { + callback.processResult(Code.OK.intValue(), path, leafNodePath); + } + } + } + }, leafNodePath); + } + + /** * Create zookeeper path recursively and optimistically. This method can throw * any of the KeeperExceptions which can be thrown by ZooKeeper#create. * KeeperException.NodeExistsException will only be thrown if the full path specified @@ -148,6 +189,23 @@ public class ZkUtils { } } + public static void deleteFullPathOptimistic(ZooKeeper zkc, String path, int znodeVersion) + throws KeeperException, InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicInteger rc = new AtomicInteger(Code.OK.intValue()); + asyncDeleteFullPathOptimistic(zkc, path, znodeVersion, new VoidCallback() { + @Override + public void processResult(int rc2, String path, Object ctx) { + rc.set(rc2); + latch.countDown(); + } + }, path); + latch.await(); + if (rc.get() != Code.OK.intValue()) { + throw KeeperException.create(Code.get(rc.get())); + } + } + private static class GetChildrenCtx { int rc; boolean done = false; http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/07852d35/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java index d962264..747398a 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java @@ -68,6 +68,7 @@ public abstract class LedgerManagerTestCase extends BookKeeperClusterTestCase { super(numBookies); activeLedgers = new SnapshotMap<Long, Boolean>(); baseConf.setLedgerManagerFactoryClass(lmFactoryCls); + baseClientConf.setLedgerManagerFactoryClass(lmFactoryCls); } public LedgerManager getLedgerManager() { http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/07852d35/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerMetadataCreationTest.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerMetadataCreationTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerMetadataCreationTest.java new file mode 100644 index 0000000..7417f99 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerMetadataCreationTest.java @@ -0,0 +1,172 @@ +/* + * + * 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.bookkeeper.meta; + +import static org.junit.Assert.assertTrue; + +import java.util.Random; +import java.util.Set; +import java.util.Vector; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.client.BookKeeper.DigestType; +import org.apache.zookeeper.ZooKeeper; +import org.junit.Assume; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class LedgerMetadataCreationTest extends LedgerManagerTestCase { + static final Logger LOG = LoggerFactory.getLogger(LedgerMetadataCreationTest.class); + + public LedgerMetadataCreationTest(Class<? extends LedgerManagerFactory> lmFactoryCls) { + super(lmFactoryCls, 4); + baseConf.setGcWaitTime(100000); + } + + @Test(timeout = 60000) + public void testLedgerCreationAndDeletionWithRandomLedgerIds() throws Exception { + testExecution(true); + } + + @Test(timeout = 60000) + public void testLedgerCreationAndDeletion() throws Exception{ + testExecution(false); + } + + public void testExecution(boolean randomLedgerId) throws Exception { + Set<Long> createRequestsLedgerIds = ConcurrentHashMap.newKeySet(); + ConcurrentLinkedDeque<Long> existingLedgerIds = new ConcurrentLinkedDeque<Long>(); + + Vector<Long> failedCreates = new Vector<Long>(); + Vector<Long> failedDeletes = new Vector<Long>(); + BookKeeper bookKeeper = new BookKeeper(baseClientConf); + + ExecutorService executor = Executors.newFixedThreadPool(300); + Random rand = new Random(); + int numberOfOperations = 20000; + for (int i = 0; i < numberOfOperations; i++) { + int iteration = i; + if (rand.nextBoolean() || existingLedgerIds.isEmpty()) { + executor.submit(() -> { + long ledgerId = -1; + try { + if (randomLedgerId) { + ledgerId = Math.abs(rand.nextLong()); + do { + if (!baseClientConf.getLedgerManagerFactoryClass() + .equals(LongHierarchicalLedgerManagerFactory.class)) { + /* + * since LongHierarchicalLedgerManager + * supports ledgerIds of decimal length upto + * 19 digits but other LedgerManagers only + * upto 10 decimals + */ + ledgerId %= 9999999999L; + } + } while (!createRequestsLedgerIds.add(ledgerId)); + } else { + ledgerId = iteration; + } + bookKeeper.createLedgerAdv(ledgerId, 3, 2, 2, DigestType.CRC32, "passwd".getBytes(), null); + existingLedgerIds.add(ledgerId); + } catch (Exception e) { + LOG.error("Got Exception while creating Ledger with ledgerId " + ledgerId, e); + failedCreates.add(ledgerId); + } + }); + } else { + executor.submit(() -> { + Long ledgerId = null; + if (rand.nextBoolean()) { + ledgerId = existingLedgerIds.pollFirst(); + } else { + ledgerId = existingLedgerIds.pollLast(); + } + if (ledgerId == null) { + return; + } + try { + bookKeeper.deleteLedger(ledgerId); + } catch (Exception e) { + LOG.error("Got Exception while deleting Ledger with ledgerId " + ledgerId, e); + failedDeletes.add(ledgerId); + } + }); + } + } + executor.shutdown(); + assertTrue("All the ledger create/delete operations should have'been completed", + executor.awaitTermination(30, TimeUnit.SECONDS)); + assertTrue("There should be no failed creates. But there are " + failedCreates.size() + " failedCreates", + failedCreates.isEmpty()); + assertTrue("There should be no failed deletes. But there are " + failedDeletes.size() + " failedDeletes", + failedDeletes.isEmpty()); + bookKeeper.close(); + } + + @Test(timeout = 60000) + public void testParentNodeDeletion() throws Exception { + /* + * run this testcase only for HierarchicalLedgerManager and + * LongHierarchicalLedgerManager, since we do recursive zNode deletes + * only for HierarchicalLedgerManager + */ + Assume.assumeTrue((baseClientConf.getLedgerManagerFactoryClass().equals(HierarchicalLedgerManagerFactory.class) + || baseClientConf.getLedgerManagerFactoryClass().equals(LongHierarchicalLedgerManagerFactory.class))); + + ZooKeeper zkc = new ZooKeeper(zkUtil.getZooKeeperConnectString(), 10000, null); + BookKeeper bookKeeper = new BookKeeper(baseClientConf); + bookKeeper.createLedgerAdv(1, 3, 2, 2, DigestType.CRC32, "passwd".getBytes(), null); + String ledgersRootPath = baseClientConf.getZkLedgersRootPath(); + String parentZnodePath; + if (baseClientConf.getLedgerManagerFactoryClass().equals(HierarchicalLedgerManagerFactory.class)) { + /* + * in HierarchicalLedgerManager (ledgersRootPath)/00/0000/L0001 + * would be the path of the znode for ledger - 1. So when ledger - 1 + * is deleted, (ledgersRootPath)/00 should also be deleted since + * there are no other children znodes + */ + parentZnodePath = ledgersRootPath + "/00"; + + } else { + /* + * in LongHierarchicalLedgerManager + * (ledgersRootPath)/000/0000/0000/0000/L0001 would be the path of + * the znode for ledger - 1. So when ledger - 1 is deleted, + * (ledgersRootPath)/000 should also be deleted since there are no + * other children znodes + */ + parentZnodePath = ledgersRootPath + "/000"; + } + assertTrue(parentZnodePath + " zNode should exist", null != zkc.exists(parentZnodePath, false)); + bookKeeper.deleteLedger(1); + assertTrue(parentZnodePath + " zNode should not exist anymore", null == zkc.exists(parentZnodePath, false)); + bookKeeper.close(); + zkc.close(); + } +} http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/07852d35/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/TestZkUtils.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/TestZkUtils.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/TestZkUtils.java new file mode 100644 index 0000000..4faf09d --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/TestZkUtils.java @@ -0,0 +1,87 @@ +/** + * + * 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.bookkeeper.util; + +import java.io.IOException; + +import org.apache.bookkeeper.test.ZooKeeperUtil; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZooDefs.Ids; +import org.apache.zookeeper.ZooKeeper; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import junit.framework.TestCase; + +public class TestZkUtils extends TestCase { + + static final Logger logger = LoggerFactory.getLogger(TestZkUtils.class); + + // ZooKeeper related variables + protected ZooKeeperUtil zkUtil = new ZooKeeperUtil(); + + @Before + @Override + public void setUp() throws Exception { + logger.info("Setting up test {}.", getName()); + zkUtil.startServer(); + } + + @After + @Override + public void tearDown() throws Exception { + zkUtil.killServer(); + logger.info("Teared down test {}.", getName()); + } + + @Test(timeout = 60000) + public void testAsyncCreateAndDeleteFullPathOptimistic() throws IOException, KeeperException, InterruptedException { + ZooKeeper zkc = new ZooKeeper(zkUtil.getZooKeeperConnectString(), 10000, null); + /* + * "/ledgers/available" is already created in ZooKeeperUtil.startServer + */ + String ledgerZnodePath = new String("/ledgers/000/000/000/001"); + ZkUtils.createFullPathOptimistic(zkc, ledgerZnodePath, "data".getBytes(), Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT); + assertTrue(ledgerZnodePath + " zNode should exist", null != zkc.exists(ledgerZnodePath, false)); + + ledgerZnodePath = new String("/ledgers/000/000/000/002"); + ZkUtils.createFullPathOptimistic(zkc, ledgerZnodePath, "data".getBytes(), Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT); + assertTrue(ledgerZnodePath + " zNode should exist", null != zkc.exists(ledgerZnodePath, false)); + + ZkUtils.deleteFullPathOptimistic(zkc, ledgerZnodePath, -1); + assertTrue(ledgerZnodePath + " zNode should not exist, since it is deleted", + null == zkc.exists(ledgerZnodePath, false)); + + ledgerZnodePath = new String("/ledgers/000/000/000/001"); + assertTrue(ledgerZnodePath + " zNode should exist", null != zkc.exists(ledgerZnodePath, false)); + ZkUtils.deleteFullPathOptimistic(zkc, ledgerZnodePath, -1); + assertTrue(ledgerZnodePath + " zNode should not exist, since it is deleted", + null == zkc.exists(ledgerZnodePath, false)); + assertTrue("/ledgers/000" + " zNode should not exist, since it should be deleted recursively", + null == zkc.exists(ledgerZnodePath, false)); + } +}
