You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by iv...@apache.org on 2011/11/28 19:25:27 UTC
svn commit: r1207495 [2/3] - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-server/conf/
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/
bookkeeper-server/src/main/java/org/apac...
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManager.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManager.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManager.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,514 @@
+package org.apache.bookkeeper.meta;
+
+/**
+ * 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.
+ */
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.HashSet;
+import java.util.List;
+
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor;
+import org.apache.bookkeeper.util.StringUtils;
+import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.AsyncCallback.StringCallback;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Hierarchical Ledger Manager which manages ledger meta in zookeeper using 2-level hierarchical znodes.
+ *
+ * <p>
+ * Hierarchical Ledger Manager first obtain a global unique id from zookeeper using a EPHEMERAL_SEQUENTIAL
+ * znode <i>(ledgersRootPath)/ledgers/idgen/ID-</i>.
+ * Since zookeeper sequential counter has a format of %10d -- that is 10 digits with 0 (zero) padding, i.e.
+ * "<path>0000000001", HierarchicalLedgerManager splits the generated id into 3 parts (2-4-4):
+ * <pre><level1 (2 digits)><level2 (4 digits)><level3 (4 digits)></pre>
+ * These 3 parts are used to form the actual ledger node path used to store ledger metadata:
+ * <pre>(ledgersRootPath)/level1/level2/L(level3)</pre>
+ * E.g Ledger 0000000001 is split into 3 parts <i>00</i>, <i>0000</i>, <i>0001</i>, which is stored in
+ * <i>(ledgersRootPath)/00/0000/L0001</i>. So each znode could have at most 10000 ledgers, which avoids
+ * failed to get children list of a too big znode during garbage collection.
+ * <p>
+ * All actived ledgers found in bookie server is managed in a sorted map, which ease us to pick
+ * up all actived ledgers belongs to (level1, level2).
+ * </p>
+ * <p>
+ * Garbage collection in HierarchicalLedgerManager is processed node by node as below:
+ * <ul>
+ * fetching all level1 nodes, by calling zk#getChildren(ledgerRootPath).
+ * <ul>
+ * for each level1 node, fetching their level2 nodes, by calling zk#getChildren(ledgerRootPath + "/" + level1)
+ * <li> fetch all existed ledgers from zookeeper in level1/level2 node, said <b>zkActiveLedgers</b>
+ * <li> fetch all active ledgers from bookie server in level1/level2, said <b>bkActiveLedgers</b>
+ * <li> loop over <b>bkActiveLedgers</b> to find those ledgers aren't existed in <b>zkActiveLedgers</b>, do garbage collection on them.
+ * </ul>
+ * </ul>
+ * Since garbage collection is running in background, HierarchicalLedgerManager did gc on single hash
+ * node at a time to avoid consuming too much resources.
+ * </p>
+ */
+class HierarchicalLedgerManager extends AbstractZkLedgerManager {
+
+ static final Logger LOG = LoggerFactory.getLogger(HierarchicalLedgerManager.class);
+ public static final String NAME = "hierarchical";
+
+ public static final int CUR_VERSION = 1;
+
+ static final String IDGENERATION_PREFIX = "/idgen/ID-";
+ private static final String MAX_ID_SUFFIX = "9999";
+ private static final String MIN_ID_SUFFIX = "0000";
+
+ // Path to generate global id
+ private final String idGenPath;
+ // A sorted map to stored all active ledger ids
+ private ConcurrentSkipListMap<Long, Boolean> activeLedgers;
+
+ // we use this to prevent long stack chains from building up in callbacks
+ ScheduledExecutorService scheduler;
+
+ /**
+ * Constructor
+ *
+ * @param conf
+ * Configuration object
+ * @param zk
+ * ZooKeeper Client Handle
+ * @param ledgerRootPath
+ * ZooKeeper Path to store ledger metadata
+ * @throws IOException when version is not compatible
+ */
+ public HierarchicalLedgerManager(AbstractConfiguration conf, ZooKeeper zk,
+ String ledgerRootPath, int layoutVersion)
+ throws IOException {
+ super(conf, zk, ledgerRootPath);
+
+ if (layoutVersion != CUR_VERSION) {
+ throw new IOException("Incompatible layout version found : "
+ + layoutVersion);
+ }
+
+ this.idGenPath = ledgerRootPath + IDGENERATION_PREFIX;
+ this.activeLedgers = new ConcurrentSkipListMap<Long, Boolean>();
+ this.scheduler = Executors.newSingleThreadScheduledExecutor();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Using HierarchicalLedgerManager with root path : " + ledgerRootPath);
+ }
+ }
+
+ @Override
+ public void close() {
+ try {
+ scheduler.shutdown();
+ } catch (Exception e) {
+ LOG.warn("Error when closing HierarchicalLedgerManager : ", e);
+ }
+ super.close();
+ }
+
+ @Override
+ public void newLedgerPath(final GenericCallback<String> ledgerCb) {
+ ZkUtils.createFullPathOptimistic(zk, idGenPath, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.EPHEMERAL_SEQUENTIAL, new StringCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx, final String idPathName) {
+ if (rc != KeeperException.Code.OK.intValue()) {
+ LOG.error("Could not generate new ledger id",
+ KeeperException.create(KeeperException.Code.get(rc), path));
+ ledgerCb.operationComplete(rc, null);
+ return;
+ }
+ /*
+ * Extract ledger id from gen path
+ */
+ long ledgerId;
+ try {
+ ledgerId = getLedgerIdFromGenPath(idPathName);
+ } catch (IOException e) {
+ LOG.error("Could not extract ledger-id from id gen path:" + path, e);
+ ledgerCb.operationComplete(KeeperException.Code.SYSTEMERROR.intValue(), null);
+ return;
+ }
+ StringCallback scb = new StringCallback() {
+ @Override
+ public void processResult(int rc, String path,
+ Object ctx, String name) {
+ if (rc != KeeperException.Code.OK.intValue()) {
+ ledgerCb.operationComplete(rc, null);
+ } else {
+ ledgerCb.operationComplete(rc, name);
+ }
+ }
+ };
+ String ledgerPath = getLedgerPath(ledgerId);
+ ZkUtils.createFullPathOptimistic(zk, ledgerPath, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, scb, null);
+ // delete the znode for id generation
+ scheduler.submit(new Runnable() {
+ @Override
+ public void run() {
+ zk.delete(idPathName, -1, new AsyncCallback.VoidCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx) {
+ if (rc != KeeperException.Code.OK.intValue()) {
+ LOG.warn("Exception during deleting znode for id generation : ",
+ KeeperException.create(KeeperException.Code.get(rc), path));
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Deleting znode for id generation : " + idPathName);
+ }
+ }
+ }
+ }, null);
+ }
+ });
+ }
+ }, null);
+ }
+
+ // get ledger id from generation path
+ private long getLedgerIdFromGenPath(String nodeName) throws IOException {
+ long ledgerId;
+ try {
+ String parts[] = nodeName.split(IDGENERATION_PREFIX);
+ ledgerId = Long.parseLong(parts[parts.length - 1]);
+ } catch (NumberFormatException e) {
+ throw new IOException(e);
+ }
+ return ledgerId;
+ }
+
+ @Override
+ public String getLedgerPath(long ledgerId) {
+ String ledgerIdStr = StringUtils.getZKStringId(ledgerId);
+ // do 2-4-4 split
+ StringBuilder sb = new StringBuilder();
+ sb.append(ledgerRootPath).append("/")
+ .append(ledgerIdStr.substring(0, 2)).append("/")
+ .append(ledgerIdStr.substring(2, 6)).append("/")
+ .append(LEDGER_NODE_PREFIX).append(ledgerIdStr.substring(6, 10));
+ return sb.toString();
+ }
+
+ @Override
+ public long getLedgerId(String pathName) throws IOException {
+ if (!pathName.startsWith(ledgerRootPath)) {
+ throw new IOException("it is not a valid hashed path name : " + pathName);
+ }
+ String hierarchicalPath = pathName.substring(ledgerRootPath.length() + 1);
+ String[] hierarchicalParts = hierarchicalPath.split("/");
+ if (hierarchicalParts.length != 3) {
+ throw new IOException("it is not a valid hierarchical path name : " + pathName);
+ }
+ hierarchicalParts[2] =
+ hierarchicalParts[2].substring(LEDGER_NODE_PREFIX.length());
+ return getLedgerId(hierarchicalParts);
+ }
+
+ // get ledger from all level nodes
+ private long getLedgerId(String...levelNodes) throws IOException {
+ try {
+ StringBuilder sb = new StringBuilder();
+ for (String node : levelNodes) {
+ sb.append(node);
+ }
+ return Long.parseLong(sb.toString());
+ } catch (NumberFormatException e) {
+ throw new IOException(e);
+ }
+ }
+
+ //
+ // Active Ledger Manager
+ //
+
+ /**
+ * Get the smallest cache id in a specified node /level1/level2
+ *
+ * @param level1
+ * 1st level node name
+ * @param level2
+ * 2nd level node name
+ * @return the smallest ledger id
+ */
+ private long getStartLedgerIdByLevel(String level1, String level2) throws IOException {
+ return getLedgerId(level1, level2, MIN_ID_SUFFIX);
+ }
+
+ /**
+ * Get the largest cache id in a specified node /level1/level2
+ *
+ * @param level1
+ * 1st level node name
+ * @param level2
+ * 2nd level node name
+ * @return the largest ledger id
+ */
+ private long getEndLedgerIdByLevel(String level1, String level2) throws IOException {
+ return getLedgerId(level1, level2, MAX_ID_SUFFIX);
+ }
+
+ @Override
+ public void asyncProcessLedgers(final Processor<Long> processor,
+ final AsyncCallback.VoidCallback finalCb, final Object context,
+ final int successRc, final int failureRc) {
+ // process 1st level nodes
+ asyncProcessLevelNodes(ledgerRootPath, new Processor<String>() {
+ @Override
+ public void process(final String l1Node, final AsyncCallback.VoidCallback cb1) {
+ if (isSpecialZnode(l1Node)) {
+ cb1.processResult(successRc, null, context);
+ return;
+ }
+ final String l1NodePath = ledgerRootPath + "/" + l1Node;
+ // process level1 path, after all children of level1 process
+ // it callback to continue processing next level1 node
+ asyncProcessLevelNodes(l1NodePath, new Processor<String>() {
+ @Override
+ public void process(String l2Node, AsyncCallback.VoidCallback cb2) {
+ // process level1/level2 path
+ String l2NodePath = ledgerRootPath + "/" + l1Node + "/" + l2Node;
+ // process each ledger
+ // after all ledger are processed, cb2 will be call to continue processing next level2 node
+ asyncProcessLedgersInSingleNode(l2NodePath, processor, cb2,
+ context, successRc, failureRc);
+ }
+ }, cb1, context, successRc, failureRc);
+ }
+ }, finalCb, context, successRc, failureRc);
+ }
+
+ /**
+ * Process hash nodes in a given path
+ */
+ private void asyncProcessLevelNodes(
+ final String path, final Processor<String> processor,
+ final AsyncCallback.VoidCallback finalCb, final Object context,
+ final int successRc, final int failureRc) {
+ zk.sync(path, new AsyncCallback.VoidCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx) {
+ if (rc != Code.OK.intValue()) {
+ LOG.error("Error syncing path " + path + " when getting its chidren: ",
+ KeeperException.create(KeeperException.Code.get(rc), path));
+ finalCb.processResult(failureRc, null, context);
+ return;
+ }
+
+ zk.getChildren(path, false, new AsyncCallback.ChildrenCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx,
+ List<String> levelNodes) {
+ if (rc != Code.OK.intValue()) {
+ LOG.error("Error polling hash nodes of " + path,
+ KeeperException.create(KeeperException.Code.get(rc), path));
+ finalCb.processResult(failureRc, null, context);
+ return;
+ }
+ AsyncListProcessor<String> listProcessor =
+ new AsyncListProcessor<String>(scheduler);
+ // process its children
+ listProcessor.process(levelNodes, processor, finalCb,
+ context, successRc, failureRc);
+ }
+ }, null);
+ }
+ }, null);
+ }
+
+ @Override
+ public void addActiveLedger(long ledgerId, boolean active) {
+ activeLedgers.put(ledgerId, active);
+ }
+
+ @Override
+ public void removeActiveLedger(long ledgerId) {
+ activeLedgers.remove(ledgerId);
+ }
+
+ @Override
+ public boolean containsActiveLedger(long ledgerId) {
+ return activeLedgers.containsKey(ledgerId);
+ }
+
+ @Override
+ public void garbageCollectLedgers(GarbageCollector gc) {
+ try {
+ List<String> l1Nodes = zk.getChildren(ledgerRootPath, null);
+ for (String l1Node : l1Nodes) {
+ if (isSpecialZnode(l1Node)) {
+ continue;
+ }
+ try {
+ List<String> l2Nodes = zk.getChildren(ledgerRootPath + "/" + l1Node, null);
+ for (String l2Node : l2Nodes) {
+ doGcByLevel(gc, l1Node, l2Node);
+ }
+ } catch (Exception e) {
+ LOG.warn("Exception during garbage collecting ledgers for " + l1Node
+ + " of " + ledgerRootPath);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Exception during garbage collecting inactive/deleted ledgers");
+ }
+ }
+
+ /**
+ * Garbage collection a single node level1/level2
+ *
+ * @param gc
+ * Garbage collector
+ * @param level1
+ * 1st level node name
+ * @param level2
+ * 2nd level node name
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ void doGcByLevel(GarbageCollector gc, final String level1, final String level2)
+ throws IOException, InterruptedException {
+
+ StringBuilder nodeBuilder = new StringBuilder();
+ nodeBuilder.append(ledgerRootPath).append("/")
+ .append(level1).append("/").append(level2);
+ String nodePath = nodeBuilder.toString();
+
+ HashSet<Long> zkActiveLedgers = getLedgersInSingleNode(nodePath);
+ // get hosted ledgers in /level1/level2
+ long startLedgerId = getStartLedgerIdByLevel(level1, level2);
+ long endLedgerId = getEndLedgerIdByLevel(level1, level2);
+ ConcurrentMap<Long, Boolean> bkActiveLedgers =
+ activeLedgers.subMap(startLedgerId, true, endLedgerId, true);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("All active ledgers from ZK for hash node "
+ + level1 + "/" + level2 + " : " + zkActiveLedgers);
+ LOG.debug("Current active ledgers from Bookie for hash node "
+ + level1 + "/" + level2 + " : " + bkActiveLedgers);
+ }
+
+ doGc(gc, bkActiveLedgers, zkActiveLedgers);
+ }
+
+ /**
+ * Do garbage collecting comparing hosted ledgers and zk ledgers
+ *
+ * @param gc
+ * Garbage collector
+ * @param bkActiveLedgers
+ * Active ledgers hosted in bookie server
+ * @param zkAllLedgers
+ * All ledgers stored in zookeeper
+ */
+ void doGc(GarbageCollector gc, ConcurrentMap<Long, Boolean> bkActiveLedgers,
+ HashSet<Long> zkAllLedgers) {
+ // remove any active ledgers that doesn't exist in zk
+ for (Long lid : bkActiveLedgers.keySet()) {
+ if (!zkAllLedgers.contains(lid)) {
+ // remove it from current active ledger
+ bkActiveLedgers.remove(lid);
+ gc.gc(lid);
+ }
+ }
+ }
+
+ /**
+ * Process list one by one in asynchronize way. Process will be stopped immediately
+ * when error occurred.
+ */
+ private static class AsyncListProcessor<T> {
+ // use this to prevent long stack chains from building up in callbacks
+ ScheduledExecutorService scheduler;
+
+ /**
+ * Constructor
+ *
+ * @param scheduler
+ * Executor used to prevent long stack chains
+ */
+ public AsyncListProcessor(ScheduledExecutorService scheduler) {
+ this.scheduler = scheduler;
+ }
+
+ /**
+ * Process list of items
+ *
+ * @param data
+ * List of data to process
+ * @param processor
+ * Callback to process element of list when success
+ * @param finalCb
+ * Final callback to be called after all elements in the list are processed
+ * @param contxt
+ * Context of final callback
+ * @param successRc
+ * RC passed to final callback on success
+ * @param failureRc
+ * RC passed to final callback on failure
+ */
+ public void process(final List<T> data, final Processor<T> processor,
+ final AsyncCallback.VoidCallback finalCb, final Object context,
+ final int successRc, final int failureRc) {
+ if (data == null || data.size() == 0) {
+ finalCb.processResult(successRc, null, context);
+ return;
+ }
+ final int size = data.size();
+ final AtomicInteger current = new AtomicInteger(0);
+ AsyncCallback.VoidCallback stubCallback = new AsyncCallback.VoidCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx) {
+ if (rc != successRc) {
+ // terminal immediately
+ finalCb.processResult(failureRc, null, context);
+ return;
+ }
+ // process next element
+ int next = current.incrementAndGet();
+ if (next >= size) { // reach the end of list
+ finalCb.processResult(successRc, null, context);
+ return;
+ }
+ final T dataToProcess = data.get(next);
+ final AsyncCallback.VoidCallback stub = this;
+ scheduler.submit(new Runnable() {
+ @Override
+ public final void run() {
+ processor.process(dataToProcess, stub);
+ }
+ });
+ }
+ };
+ T firstElement = data.get(0);
+ processor.process(firstElement, stubCallback);
+ }
+ }
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,204 @@
+package org.apache.bookkeeper.meta;
+
+/**
+ * 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.
+ */
+
+import java.io.IOException;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.ZooKeeper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.configuration.ConfigurationException;
+
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.meta.FlatLedgerManager;
+
+/**
+ * This class encapsulates ledger layout information that is persistently stored
+ * in zookeeper. It provides parsing and serialization methods of such information.
+ *
+ */
+class LedgerLayout {
+ static final Logger LOG = LoggerFactory.getLogger(LedgerLayout.class);
+
+ // Znode name to store layout information
+ public static final String LAYOUT_ZNODE = "LAYOUT";
+ // version of ledger layout metadata
+ public static final int LAYOUT_FORMAT_VERSION = 1;
+
+ /**
+ * Read ledger layout from zookeeper
+ *
+ * @param zk ZooKeeper Client
+ * @param ledgersRoot Root of the ledger namespace to check
+ * @return ledger layout, or null if none set in zookeeper
+ */
+ public static LedgerLayout readLayout(final ZooKeeper zk, final String ledgersRoot)
+ throws IOException, KeeperException {
+ String ledgersLayout = ledgersRoot + "/" + LAYOUT_ZNODE;
+
+ try {
+ LedgerLayout layout;
+
+ try {
+ byte[] layoutData = zk.getData(ledgersLayout, false, null);
+ layout = parseLayout(layoutData);
+ } catch (KeeperException.NoNodeException nne) {
+ return null;
+ }
+
+ return layout;
+ } catch (InterruptedException ie) {
+ throw new IOException(ie);
+ }
+ }
+
+ static final String splitter = ":";
+ static final String lSplitter = "\n";
+
+ // ledger manager class
+ private String managerType;
+ // ledger manager version
+ private int managerVersion;
+
+ // layout version of how to store layout information
+ private int layoutFormatVersion = LAYOUT_FORMAT_VERSION;
+
+ /**
+ * Ledger Layout Constructor
+ *
+ * @param type
+ * Ledger Manager Type
+ * @param managerVersion
+ * Ledger Manager Version
+ * @param layoutFormatVersion
+ * Ledger Layout Format Version
+ */
+ public LedgerLayout(String managerType, int managerVersion) {
+ this.managerType = managerType;
+ this.managerVersion = managerVersion;
+ }
+
+ public String getManagerType() {
+ return this.managerType;
+ }
+
+ public int getManagerVersion() {
+ return this.managerVersion;
+ }
+
+ /**
+ * Store the ledger layout into zookeeper
+ */
+ public void store(final ZooKeeper zk, String ledgersRoot)
+ throws IOException, KeeperException, InterruptedException {
+ String ledgersLayout = ledgersRoot + "/" + LAYOUT_ZNODE;
+ zk.create(ledgersLayout, serialize(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ }
+
+ /**
+ * Generates a byte array based on the LedgerLayout object.
+ *
+ * @return byte[]
+ */
+ private byte[] serialize() throws IOException {
+ StringBuilder sb = new StringBuilder();
+ sb.append(layoutFormatVersion).append(lSplitter)
+ .append(managerType).append(splitter).append(managerVersion);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Serialized layout info: " + sb.toString());
+ }
+
+ return sb.toString().getBytes("UTF-8");
+ }
+
+ /**
+ * Parses a given byte array and transforms into a LedgerLayout object
+ *
+ * @param bytes
+ * byte array to parse
+ * @param znodeVersion
+ * version of znode
+ * @return LedgerLayout
+ * @throws IOException
+ * if the given byte[] cannot be parsed
+ */
+ private static LedgerLayout parseLayout(byte[] bytes) throws IOException {
+ String layout = new String(bytes, "UTF-8");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Parsing Layout: " + layout);
+ }
+
+ String lines[] = layout.split(lSplitter);
+
+ try {
+ int layoutFormatVersion = new Integer(lines[0]);
+ if (LAYOUT_FORMAT_VERSION != layoutFormatVersion) {
+ throw new IOException("Metadata version not compatible. Expected "
+ + LAYOUT_FORMAT_VERSION + ", but got " + layoutFormatVersion);
+ }
+
+ if (lines.length < 2) {
+ throw new IOException("Ledger manager and its version absent from layout: " + layout);
+ }
+
+ String[] parts = lines[1].split(splitter);
+ if (parts.length != 2) {
+ throw new IOException("Invalid Ledger Manager defined in layout : " + layout);
+ }
+ // ledger manager class
+ String managerType = parts[0];
+ // ledger manager version
+ int managerVersion = new Integer(parts[1]);
+ return new LedgerLayout(managerType, managerVersion);
+ } catch (NumberFormatException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (null == obj) {
+ return false;
+ }
+ if (!(obj instanceof LedgerLayout)) {
+ return false;
+ }
+ LedgerLayout other = (LedgerLayout)obj;
+ return managerType.equals(other.managerType) &&
+ managerVersion == other.managerVersion;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("LV").append(layoutFormatVersion).append(":")
+ .append(",Type:").append(managerType).append(":")
+ .append(managerVersion);
+ return sb.toString();
+ }
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,144 @@
+package org.apache.bookkeeper.meta;
+
+/**
+ * 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.
+ */
+
+import java.io.IOException;
+
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor;
+
+/**
+ * LedgerManager takes responsibility of ledger management
+ *
+ * <ul>
+ * <li>How to store ledger meta (e.g. in ZooKeeper or other key/value store)
+ * <li>How to manager active ledgers (so know how to do garbage collection)
+ * <li>How to garbage collect inactive/deleted ledgers
+ * </ul>
+ */
+public interface LedgerManager {
+
+ /**
+ * Get the path that is used to store ledger metadata
+ *
+ * @param ledgerId
+ * Ledger ID
+ * @return ledger node path
+ */
+ public String getLedgerPath(long ledgerId);
+
+ /**
+ * Get ledger id from its ledger path
+ *
+ * @param ledgerPath
+ * Ledger path to store metadata
+ * @return ledger id
+ * @throws IOException when the ledger path is invalid
+ */
+ public long getLedgerId(String ledgerPath) throws IOException;
+
+ /**
+ * Create a new zk ledger path.
+ *
+ * @param cb
+ * Callback when getting new zk ledger path to create.
+ */
+ public abstract void newLedgerPath(GenericCallback<String> cb);
+
+ /**
+ * Loop to process all ledgers.
+ * <p>
+ * <ul>
+ * After all ledgers were processed, finalCb will be triggerred:
+ * <li> if all ledgers are processed done with OK, success rc will be passed to finalCb.
+ * <li> if some ledgers are prcoessed failed, failure rc will be passed to finalCb.
+ * </ul>
+ * </p>
+ *
+ * @param processor
+ * Ledger Processor to process a specific ledger
+ * @param finalCb
+ * Callback triggered after all ledgers are processed
+ * @param context
+ * Context of final callback
+ * @param successRc
+ * Success RC code passed to finalCb when callback
+ * @param failureRc
+ * Failure RC code passed to finalCb when exceptions occured.
+ */
+ public void asyncProcessLedgers(Processor<Long> processor, AsyncCallback.VoidCallback finalCb,
+ Object context, int successRc, int failureRc);
+
+ /**
+ * Add active ledger
+ *
+ * @param ledgerId
+ * Ledger ID
+ * @param active
+ * Status of ledger
+ */
+ public void addActiveLedger(long ledgerId, boolean active);
+
+ /**
+ * Remove active ledger
+ *
+ * @param ledgerId
+ * Ledger ID
+ */
+ public void removeActiveLedger(long ledgerId);
+
+ /**
+ * Is Ledger ledgerId in active ledgers set
+ *
+ * @param ledgerId
+ * Ledger ID
+ * @return true if the ledger is in active ledgers set, otherwise return false
+ */
+ public boolean containsActiveLedger(long ledgerId);
+
+ /**
+ * Garbage Collector which handles ledger deletion in server side
+ */
+ public static interface GarbageCollector {
+ /**
+ * garbage collecting a specific ledger
+ *
+ * @param ledgerId
+ * Ledger ID to be garbage collected
+ */
+ public void gc(long ledgerId);
+ }
+
+ /**
+ * Garbage collecting all inactive/deleted ledgers
+ * <p>
+ * GarbageCollector#gc is triggered each time we found a ledger could be garbage collected.
+ * After method finished, all those inactive ledgers should be garbage collected.
+ * </p>
+ *
+ * @param gc garbage collector
+ */
+ public void garbageCollectLedgers(GarbageCollector gc);
+
+ /**
+ * Close ledger manager
+ */
+ public void close();
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,98 @@
+package org.apache.bookkeeper.meta;
+
+/**
+ * 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.
+ */
+
+import java.io.IOException;
+
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * <code>LedgerManagerFactory</code> takes responsibility of creating new ledger manager.
+ */
+public class LedgerManagerFactory {
+ /**
+ * Create new Ledger Manager.
+ *
+ * @param conf
+ * Configuration Object.
+ * @param zk
+ * ZooKeeper Client Handle, talk to zk to know which ledger manager is used.
+ * @return new ledger manager
+ * @throws IOException
+ */
+ public static LedgerManager newLedgerManager(
+ final AbstractConfiguration conf, final ZooKeeper zk)
+ throws IOException, KeeperException, InterruptedException {
+ String lmType = conf.getLedgerManagerType();
+ String ledgerRootPath = conf.getZkLedgersRootPath();
+
+ if (null == ledgerRootPath || ledgerRootPath.length() == 0) {
+ throw new IOException("Empty Ledger Root Path.");
+ }
+
+ // if zk is null, return the default ledger manager
+ if (zk == null) {
+ return new FlatLedgerManager(conf, zk,
+ ledgerRootPath, FlatLedgerManager.CUR_VERSION);
+ }
+
+ // check that the configured ledger manager is
+ // compatible with the existing layout
+ LedgerLayout layout = LedgerLayout.readLayout(zk, ledgerRootPath);
+ if (layout == null) { // no existing layout
+ if (lmType == null
+ || lmType.equals(FlatLedgerManager.NAME)) {
+ layout = new LedgerLayout(FlatLedgerManager.NAME,
+ FlatLedgerManager.CUR_VERSION);
+ } else if (lmType.equals(HierarchicalLedgerManager.NAME)) {
+ layout = new LedgerLayout(HierarchicalLedgerManager.NAME,
+ HierarchicalLedgerManager.CUR_VERSION);
+ } else {
+ throw new IOException("Unknown ledger manager type " + lmType);
+ }
+ try {
+ layout.store(zk, ledgerRootPath);
+ } catch (KeeperException.NodeExistsException nee) {
+ LedgerLayout layout2 = LedgerLayout.readLayout(zk, ledgerRootPath);
+ if (!layout2.equals(layout)) {
+ throw new IOException("Contention writing to layout to zookeeper, "
+ + " other layout " + layout2 + " is incompatible with our "
+ + "layout " + layout);
+ }
+ }
+ } else if (lmType != null && !layout.getManagerType().equals(lmType)) {
+ throw new IOException("Configured layout " + lmType
+ + " does not match existing layout " + layout.getManagerType());
+ }
+
+ // create the ledger manager
+ if (FlatLedgerManager.NAME.equals(layout.getManagerType())) {
+ return new FlatLedgerManager(conf, zk, ledgerRootPath,
+ layout.getManagerVersion());
+ } else if (HierarchicalLedgerManager.NAME.equals(layout.getManagerType())) {
+ return new HierarchicalLedgerManager(conf, zk, ledgerRootPath,
+ layout.getManagerVersion());
+ } else {
+ throw new IOException("Unknown ledger manager type: " + lmType);
+ }
+ }
+
+}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java Mon Nov 28 18:25:18 2011
@@ -29,6 +29,8 @@ import java.net.MalformedURLException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
+import org.apache.zookeeper.KeeperException;
+
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -56,7 +58,8 @@ public class BookieServer implements NIO
DeathWatcher deathWatcher;
static Logger LOG = LoggerFactory.getLogger(BookieServer.class);
- public BookieServer(ServerConfiguration conf) throws IOException {
+ public BookieServer(ServerConfiguration conf)
+ throws IOException, KeeperException, InterruptedException {
this.conf = conf;
this.bookie = new Bookie(conf);
}
@@ -217,8 +220,8 @@ public class BookieServer implements NIO
* @throws IOException
* @throws InterruptedException
*/
- public static void main(String[] args) throws IOException, InterruptedException,
- IllegalArgumentException {
+ public static void main(String[] args)
+ throws IOException, KeeperException, InterruptedException {
ServerConfiguration conf = null;
try {
conf = parseArgs(args);
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java Mon Nov 28 18:25:18 2011
@@ -22,7 +22,13 @@
package org.apache.bookkeeper.proto;
import java.net.InetSocketAddress;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.zookeeper.AsyncCallback;
import org.jboss.netty.buffer.ChannelBuffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Declaration of a callback interfaces used in bookkeeper client library but
@@ -30,6 +36,9 @@ import org.jboss.netty.buffer.ChannelBuf
*/
public class BookkeeperInternalCallbacks {
+
+ static final Logger LOG = LoggerFactory.getLogger(BookkeeperInternalCallbacks.class);
+
/**
* Callback for calls from BookieClient objects. Such calls are for replies
* of write operations (operations to add an entry to a ledger).
@@ -54,4 +63,70 @@ public class BookkeeperInternalCallbacks
public interface ReadEntryCallback {
void readEntryComplete(int rc, long ledgerId, long entryId, ChannelBuffer buffer, Object ctx);
}
+
+ /**
+ * This is a multi callback object that waits for all of
+ * the multiple async operations to complete. If any fail, then we invoke
+ * the final callback with a provided failureRc
+ */
+ public static class MultiCallback implements AsyncCallback.VoidCallback {
+ // Number of expected callbacks
+ final int expected;
+ final int failureRc;
+ final int successRc;
+ // Final callback and the corresponding context to invoke
+ final AsyncCallback.VoidCallback cb;
+ final Object context;
+ // This keeps track of how many operations have completed
+ final AtomicInteger done = new AtomicInteger();
+ // List of the exceptions from operations that completed unsuccessfully
+ final LinkedBlockingQueue<Integer> exceptions = new LinkedBlockingQueue<Integer>();
+
+ public MultiCallback(int expected, AsyncCallback.VoidCallback cb, Object context, int successRc, int failureRc) {
+ this.expected = expected;
+ this.cb = cb;
+ this.context = context;
+ this.failureRc = failureRc;
+ this.successRc = successRc;
+ if (expected == 0) {
+ cb.processResult(successRc, null, context);
+ }
+ }
+
+ private void tick() {
+ if (done.incrementAndGet() == expected) {
+ if (exceptions.isEmpty()) {
+ cb.processResult(successRc, null, context);
+ } else {
+ cb.processResult(failureRc, null, context);
+ }
+ }
+ }
+
+ @Override
+ public void processResult(int rc, String path, Object ctx) {
+ if (rc != successRc) {
+ LOG.error("Error in mutil callback : " + rc);
+ exceptions.add(rc);
+ }
+ tick();
+ }
+
+ }
+
+ /**
+ * Processor to process a specific element
+ */
+ public static interface Processor<T> {
+ /**
+ * Process a specific element
+ *
+ * @param data
+ * data to process
+ * @param iterationCallback
+ * Callback to invoke when process has been done.
+ */
+ public void process(T data, AsyncCallback.VoidCallback cb);
+ }
+
}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/LocalBookKeeper.java Mon Nov 28 18:25:18 2011
@@ -122,7 +122,8 @@ public class LocalBookKeeper {
LOG.error("Exception while creating znodes", e);
}
}
- private void runBookies(ServerConfiguration baseConf) throws IOException {
+ private void runBookies(ServerConfiguration baseConf)
+ throws IOException, KeeperException, InterruptedException {
LOG.info("Starting Bookie(s)");
// Create Bookie Servers (B1, B2, B3)
@@ -148,7 +149,8 @@ public class LocalBookKeeper {
}
}
- public static void main(String[] args) throws IOException, InterruptedException {
+ public static void main(String[] args)
+ throws IOException, KeeperException, InterruptedException {
if(args.length < 1) {
usage();
System.exit(-1);
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/Main.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/Main.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/Main.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/Main.java Mon Nov 28 18:25:18 2011
@@ -37,7 +37,7 @@ public class Main {
* @throws InterruptedException
* @throws IOException
*/
- public static void main(String[] args) throws IOException, InterruptedException {
+ public static void main(String[] args) throws Exception {
if (args.length < 1 || !(args[0].equals("client") || args[0].equals("bookie"))) {
usage();
return;
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java Mon Nov 28 18:25:18 2011
@@ -28,11 +28,6 @@ import java.net.InetSocketAddress;
*/
public class StringUtils {
- /*
- * Path to ledger metadata. ZooKeeper appends a sequence number to L.
- */
- static public final String prefix = "/ledgers/L";
-
/**
* Parses address into IP and port.
*
@@ -71,24 +66,4 @@ public class StringUtils {
return String.format("%010d", id);
}
- /**
- * Get the path for the ledger metadata node
- *
- * @return
- */
- public static String getLedgerNodePath(long ledgerId) {
- return prefix + StringUtils.getZKStringId(ledgerId);
- }
-
- public static long getLedgerId(String nodeName) throws IOException {
- long ledgerId;
- try {
- String parts[] = nodeName.split(prefix);
- ledgerId = Long.parseLong(parts[parts.length - 1]);
- } catch (NumberFormatException e) {
- throw new IOException(e);
- }
- return ledgerId;
- }
-
}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/ZkUtils.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,94 @@
+/*
+ *
+ * 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.File;
+import java.util.List;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.AsyncCallback.StringCallback;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * Provided utilites for zookeeper access, etc.
+ */
+public class ZkUtils {
+
+ /**
+ * Create zookeeper path recursively
+ *
+ * @param zk
+ * Zookeeper client
+ * @param originalPath
+ * Zookeeper full path
+ * @param data
+ * Zookeeper data
+ * @param acl
+ * Acl of the zk path
+ * @param createMode
+ * Create mode of zk path
+ * @param callback
+ * Callback
+ * @param ctx
+ * Context object
+ */
+ public static void createFullPathOptimistic(
+ final ZooKeeper zk, final String originalPath, final byte[] data,
+ final List<ACL> acl, final CreateMode createMode,
+ final AsyncCallback.StringCallback callback, final Object ctx) {
+
+ zk.create(originalPath, data, acl, createMode, new StringCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx, String name) {
+
+ if (rc != Code.NONODE.intValue()) {
+ callback.processResult(rc, path, ctx, name);
+ return;
+ }
+
+ // Since I got a nonode, it means that my parents don't exist
+ // create mode is persistent since ephemeral nodes can't be
+ // parents
+ createFullPathOptimistic(zk, new File(originalPath).getParent().replace("\\", "/"), new byte[0], acl,
+ CreateMode.PERSISTENT, new StringCallback() {
+
+ @Override
+ public void processResult(int rc, String path, Object ctx, String name) {
+ if (rc == Code.OK.intValue() || rc == Code.NODEEXISTS.intValue()) {
+ // succeeded in creating the parent, now
+ // create the original path
+ createFullPathOptimistic(zk, originalPath, data, acl, createMode, callback,
+ ctx);
+ } else {
+ callback.processResult(rc, path, ctx, name);
+ }
+ }
+ }, ctx);
+ }
+ }, ctx);
+
+ }
+
+}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java Mon Nov 28 18:25:18 2011
@@ -38,7 +38,6 @@ import org.jboss.netty.buffer.ChannelBuf
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.util.StringUtils;
import org.apache.bookkeeper.test.BaseTestCase;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -56,8 +55,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -214,8 +211,8 @@ public class BookieRecoveryTest extends
/**
* Helper method to verify that we can read the recovered ledger entries.
*
- * @param numLedgers
- * Number of ledgers to verify
+ * @param oldLhs
+ * Old Ledger Handles
* @param startEntryId
* Start Entry Id to read
* @param endEntryId
@@ -223,12 +220,12 @@ public class BookieRecoveryTest extends
* @throws BKException
* @throws InterruptedException
*/
- private void verifyRecoveredLedgers(int numLedgers, long startEntryId, long endEntryId) throws BKException,
+ private void verifyRecoveredLedgers(List<LedgerHandle> oldLhs, long startEntryId, long endEntryId) throws BKException,
InterruptedException {
// Get a set of LedgerHandles for all of the ledgers to verify
List<LedgerHandle> lhs = new ArrayList<LedgerHandle>();
- for (int i = 0; i < numLedgers; i++) {
- lhs.add(bkc.openLedger(i + 1, digestType, baseClientConf.getBookieRecoveryPasswd()));
+ for (int i = 0; i < oldLhs.size(); i++) {
+ lhs.add(bkc.openLedger(oldLhs.get(i).getId(), digestType, baseClientConf.getBookieRecoveryPasswd()));
}
// Read the ledger entries to verify that they are all present and
// correct in the new bookie.
@@ -293,7 +290,7 @@ public class BookieRecoveryTest extends
}
// Verify the recovered ledger entries are okay.
- verifyRecoveredLedgers(numLedgers, 0, 2 * numMsgs - 1);
+ verifyRecoveredLedgers(lhs, 0, 2 * numMsgs - 1);
}
/**
@@ -349,7 +346,7 @@ public class BookieRecoveryTest extends
}
// Verify the recovered ledger entries are okay.
- verifyRecoveredLedgers(numLedgers, 0, 2 * numMsgs - 1);
+ verifyRecoveredLedgers(lhs, 0, 2 * numMsgs - 1);
}
/**
@@ -392,7 +389,7 @@ public class BookieRecoveryTest extends
bkAdmin.recoverBookieData(bookieSrc, bookieDest);
// Verify the recovered ledger entries are okay.
- verifyRecoveredLedgers(numLedgers, 0, 2 * numMsgs - 1);
+ verifyRecoveredLedgers(lhs, 0, 2 * numMsgs - 1);
}
/**
@@ -438,7 +435,7 @@ public class BookieRecoveryTest extends
bkAdmin.recoverBookieData(bookieSrc, bookieDest);
// Verify the recovered ledger entries are okay.
- verifyRecoveredLedgers(numLedgers, 0, 2 * numMsgs - 1);
+ verifyRecoveredLedgers(lhs, 0, 2 * numMsgs - 1);
}
private static class ReplicationVerificationCallback implements ReadEntryCallback {
@@ -472,7 +469,7 @@ public class BookieRecoveryTest extends
}
private boolean verifyFullyReplicated(LedgerHandle lh, long untilEntry) throws Exception {
- String znodepath = StringUtils.getLedgerNodePath(lh.getId());
+ String znodepath = bkc.getLedgerManager().getLedgerPath(lh.getId());
Stat stat = bkc.getZkHandle().exists(znodepath, false);
assertNotNull(stat);
byte[] mdbytes = bkc.getZkHandle().getData(znodepath, false, stat);
@@ -519,7 +516,7 @@ public class BookieRecoveryTest extends
private boolean findDupesInEnsembles(List<LedgerHandle> lhs) throws Exception {
long numDupes = 0;
for (LedgerHandle lh : lhs) {
- String znodepath = StringUtils.getLedgerNodePath(lh.getId());
+ String znodepath = bkc.getLedgerManager().getLedgerPath(lh.getId());
Stat stat = bkc.getZkHandle().exists(znodepath, false);
assertNotNull(stat);
byte[] mdbytes = bkc.getZkHandle().getData(znodepath, false, stat);
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerCacheTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerCacheTest.java?rev=1207495&r1=1207494&r2=1207495&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerCacheTest.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerCacheTest.java Mon Nov 28 18:25:18 2011
@@ -63,7 +63,7 @@ public class LedgerCacheTest extends Tes
@Override
@Before
- public void setUp() throws IOException {
+ public void setUp() throws Exception {
String txnDirName = System.getProperty("txnDir");
if (txnDirName != null) {
txnDir = new File(txnDirName);
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalAsyncLedgerOpsTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalAsyncLedgerOpsTest.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalAsyncLedgerOpsTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalAsyncLedgerOpsTest.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,47 @@
+package org.apache.bookkeeper.meta;
+
+/*
+ *
+ * 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.
+ *
+ */
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManager;
+import org.apache.bookkeeper.test.AsyncLedgerOpsTest;
+
+import org.junit.Before;
+
+/**
+ * Test ledger delete using HierarchicalLedgerManager
+ */
+public class HierarchicalAsyncLedgerOpsTest extends AsyncLedgerOpsTest {
+
+ public HierarchicalAsyncLedgerOpsTest(DigestType digestType) {
+ super(digestType);
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ baseConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ baseClientConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ super.setUp();
+ }
+
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieFailureTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieFailureTest.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieFailureTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieFailureTest.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,47 @@
+package org.apache.bookkeeper.meta;
+
+/*
+ *
+ * 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.
+ *
+ */
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManager;
+import org.apache.bookkeeper.test.BookieFailureTest;
+
+import org.junit.Before;
+
+/**
+ * Test Bookie Failure using HierarchicalLedgerManager
+ */
+public class HierarchicalBookieFailureTest extends BookieFailureTest {
+
+ public HierarchicalBookieFailureTest(DigestType digestType) {
+ super(digestType);
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ baseConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ baseClientConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ super.setUp();
+ }
+
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieReadWriteTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieReadWriteTest.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieReadWriteTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieReadWriteTest.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,47 @@
+package org.apache.bookkeeper.meta;
+
+/*
+*
+* 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.
+*
+*/
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManager;
+import org.apache.bookkeeper.test.BookieReadWriteTest;
+
+import org.junit.Before;
+
+/**
+* Test read / write using HierarchicalLedgerManager
+*/
+public class HierarchicalBookieReadWriteTest extends BookieReadWriteTest {
+
+ public HierarchicalBookieReadWriteTest(DigestType digestType) {
+ super(digestType);
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ baseConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ baseClientConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ super.setUp();
+ }
+
+}
\ No newline at end of file
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieRecoveryTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieRecoveryTest.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieRecoveryTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalBookieRecoveryTest.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,46 @@
+package org.apache.bookkeeper.meta;
+
+/*
+ *
+ * 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.
+ *
+ */
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.BookieRecoveryTest;
+
+import org.junit.Before;
+
+/**
+ * Test Bookie Recovery using HierarchicalLedgerManager
+ */
+public class HierarchicalBookieRecoveryTest extends BookieRecoveryTest {
+
+ public HierarchicalBookieRecoveryTest(DigestType digestType) {
+ super(digestType);
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ baseConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ baseClientConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ super.setUp();
+ }
+
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalLedgerDeleteTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalLedgerDeleteTest.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalLedgerDeleteTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/HierarchicalLedgerDeleteTest.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,47 @@
+package org.apache.bookkeeper.meta;
+
+/*
+ *
+ * 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.
+ *
+ */
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManager;
+import org.apache.bookkeeper.test.LedgerDeleteTest;
+
+import org.junit.Before;
+
+/**
+ * Test ledger delete using HierarchicalLedgerManager
+ */
+public class HierarchicalLedgerDeleteTest extends LedgerDeleteTest {
+
+ public HierarchicalLedgerDeleteTest(DigestType digestType) {
+ super(digestType);
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ baseConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ baseClientConf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ super.setUp();
+ }
+
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,136 @@
+/*
+ *
+ * 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 java.io.IOException;
+import java.util.Random;
+import java.lang.reflect.Field;
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.bookkeeper.test.BaseTestCase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.Assert.*;
+
+public class LedgerLayoutTest extends BaseTestCase {
+ DigestType digestType;
+
+ public LedgerLayoutTest(DigestType digestType) {
+ super(0);
+ this.digestType = digestType;
+ }
+
+ @Test
+ public void testLedgerLayout() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ String ledgerRootPath = "/testLedgerLayout";
+
+ zkc.create(ledgerRootPath, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ LedgerLayout layout = LedgerLayout.readLayout(zkc, ledgerRootPath);
+ assertTrue("Layout should be null", layout == null);
+
+ String testName = "foobar";
+ int testVersion = 0xdeadbeef;
+ // use layout defined in configuration also create it in zookeeper
+ LedgerLayout layout2 = new LedgerLayout(testName, testVersion);
+ layout2.store(zkc, ledgerRootPath);
+
+ layout = LedgerLayout.readLayout(zkc, ledgerRootPath);
+ assertEquals(testName, layout.getManagerType());
+ assertEquals(testVersion, layout.getManagerVersion());
+ }
+
+ private void writeLedgerLayout(
+ String ledgersRootPath,
+ String managerType,
+ int managerVersion, int layoutVersion)
+ throws Exception {
+ LedgerLayout layout = new LedgerLayout(managerType, managerVersion);
+
+ Field f = LedgerLayout.class.getDeclaredField("layoutFormatVersion");
+ f.setAccessible(true);
+ f.set(layout, layoutVersion);
+
+ layout.store(zkc, ledgersRootPath);
+ }
+
+ @Test
+ public void testBadVersionLedgerLayout() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration();
+ // write bad version ledger layout
+ writeLedgerLayout(conf.getZkLedgersRootPath(), FlatLedgerManager.NAME,
+ FlatLedgerManager.CUR_VERSION, LedgerLayout.LAYOUT_FORMAT_VERSION + 1);
+
+ try {
+ LedgerLayout.readLayout(zkc, conf.getZkLedgersRootPath());
+ fail("Shouldn't reach here!");
+ } catch (IOException ie) {
+ assertTrue("Invalid exception", ie.getMessage().contains("version not compatible"));
+ }
+ }
+
+ @Test
+ public void testAbsentLedgerManagerLayout() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration();
+ String ledgersLayout = conf.getZkLedgersRootPath() + "/" + LedgerLayout.LAYOUT_ZNODE;
+ // write bad format ledger layout
+ StringBuilder sb = new StringBuilder();
+ sb.append(LedgerLayout.LAYOUT_FORMAT_VERSION).append("\n");
+ zkc.create(ledgersLayout, sb.toString().getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ try {
+ LedgerLayout.readLayout(zkc, conf.getZkLedgersRootPath());
+ fail("Shouldn't reach here!");
+ } catch (IOException ie) {
+ assertTrue("Invalid exception", ie.getMessage().contains("version absent from"));
+ }
+ }
+
+ @Test
+ public void testBaseLedgerManagerLayout() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration();
+ String rootPath = conf.getZkLedgersRootPath();
+ String ledgersLayout = rootPath + "/" + LedgerLayout.LAYOUT_ZNODE;
+ // write bad format ledger layout
+ StringBuilder sb = new StringBuilder();
+ sb.append(LedgerLayout.LAYOUT_FORMAT_VERSION).append("\n")
+ .append(FlatLedgerManager.NAME);
+ zkc.create(ledgersLayout, sb.toString().getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ try {
+ LedgerLayout.readLayout(zkc, rootPath);
+ fail("Shouldn't reach here!");
+ } catch (IOException ie) {
+ assertTrue("Invalid exception", ie.getMessage().contains("Invalid Ledger Manager"));
+ }
+ }
+}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java?rev=1207495&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java Mon Nov 28 18:25:18 2011
@@ -0,0 +1,261 @@
+/*
+ *
+ * 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 org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs.Ids;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.CountDownLatch;
+
+import java.util.List;
+import java.util.ArrayList;
+
+import org.apache.bookkeeper.test.BaseTestCase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.Assert.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestLedgerManager extends BaseTestCase {
+ static Logger LOG = LoggerFactory.getLogger(TestLedgerManager.class);
+
+ public TestLedgerManager(DigestType digestType) {
+ super(0);
+ }
+
+ /**
+ * Test bad client configuration
+ */
+ @Test
+ public void testBadConf() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration();
+
+ // success case
+ String root0 = "/goodconf0";
+ zkc.create(root0, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ conf.setZkLedgersRootPath(root0);
+
+ LedgerManager m = LedgerManagerFactory.newLedgerManager(conf, zkc);
+ assertTrue("Ledger manager is unexpected type",
+ (m instanceof FlatLedgerManager));
+
+ // mismatching conf
+ conf.setLedgerManagerType(HierarchicalLedgerManager.NAME);
+ try {
+ LedgerManagerFactory.newLedgerManager(conf, zkc);
+ fail("Shouldn't reach here");
+ } catch (Exception e) {
+ assertTrue("Invalid exception",
+ e.getMessage().contains("does not match existing layout"));
+ }
+
+ // invalid ledger manager
+ String root1 = "/badconf1";
+ zkc.create(root1, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ conf.setZkLedgersRootPath(root1);
+
+ conf.setLedgerManagerType("DoesNotExist");
+ try {
+ LedgerManagerFactory.newLedgerManager(conf, zkc);
+ fail("Shouldn't reach here");
+ } catch (Exception e) {
+ assertTrue("Invalid exception",
+ e.getMessage().contains("Unknown ledger manager type "));
+ }
+ }
+
+ /**
+ * Test bad zk configuration
+ */
+ @Test
+ public void testBadZkContents() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration();
+
+ // bad type in zookeeper
+ String root0 = "/badzk0";
+ zkc.create(root0, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ conf.setZkLedgersRootPath(root0);
+
+ new LedgerLayout("DoesNotExist",
+ 0xdeadbeef).store(zkc, root0);
+
+ try {
+ LedgerManagerFactory.newLedgerManager(conf, zkc);
+ fail("Shouldn't reach here");
+ } catch (Exception e) {
+ assertTrue("Invalid exception",
+ e.getMessage().contains("Unknown ledger manager type"));
+ }
+
+ // bad version in zookeeper
+ String root1 = "/badzk1";
+ zkc.create(root1, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ conf.setZkLedgersRootPath(root1);
+
+ new LedgerLayout(FlatLedgerManager.NAME,
+ 0xdeadbeef).store(zkc, root1);
+
+ try {
+ LedgerManagerFactory.newLedgerManager(conf, zkc);
+ fail("Shouldn't reach here");
+ } catch (Exception e) {
+ assertTrue("Invalid exception",
+ e.getMessage().contains("Incompatible layout version found"));
+ }
+ }
+
+ private static class CreateLMThread extends Thread {
+ private boolean success = false;
+ private final String type;
+ private final String root;
+ private final CyclicBarrier barrier;
+ private ZooKeeper zkc;
+
+ CreateLMThread(String root, String type, CyclicBarrier barrier) throws Exception {
+ this.type = type;
+ this.barrier = barrier;
+ this.root = root;
+ final CountDownLatch latch = new CountDownLatch(1);
+ zkc = new ZooKeeper("127.0.0.1", 10000, new Watcher() {
+ public void process(WatchedEvent event) {
+ latch.countDown();
+ }
+ });
+ latch.await();
+ }
+
+ public void run() {
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setLedgerManagerType(type);
+
+ try {
+ barrier.await();
+ LedgerManagerFactory.newLedgerManager(conf, zkc);
+
+ success = true;
+ } catch (Exception e) {
+ LOG.error("Failed to create ledger manager", e);
+ }
+ }
+
+ public boolean isSuccessful() {
+ return success;
+ }
+
+ public void close() throws Exception {
+ zkc.close();
+ }
+ }
+
+ // test concurrent
+ @Test
+ public void testConcurrent1() throws Exception {
+ /// everyone creates the same
+ int numThreads = 50;
+
+ // bad version in zookeeper
+ String root0 = "/lmroot0";
+ zkc.create(root0, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ CyclicBarrier barrier = new CyclicBarrier(numThreads+1);
+ List<CreateLMThread> threads = new ArrayList<CreateLMThread>(numThreads);
+ for (int i = 0; i < numThreads; i++) {
+ CreateLMThread t = new CreateLMThread(root0, FlatLedgerManager.NAME, barrier);
+ t.start();
+ threads.add(t);
+ }
+
+ barrier.await();
+
+ boolean success = true;
+ for (CreateLMThread t : threads) {
+ t.join();
+ t.close();
+ success = t.isSuccessful() && success;
+ }
+ assertTrue("Not all ledger managers created", success);
+ }
+
+ @Test
+ public void testConcurrent2() throws Exception {
+ /// odd create different
+ int numThreadsEach = 25;
+
+ // bad version in zookeeper
+ String root0 = "/lmroot0";
+ zkc.create(root0, new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ CyclicBarrier barrier = new CyclicBarrier(numThreadsEach*2+1);
+ List<CreateLMThread> threadsA = new ArrayList<CreateLMThread>(numThreadsEach);
+ for (int i = 0; i < numThreadsEach; i++) {
+ CreateLMThread t = new CreateLMThread(root0, FlatLedgerManager.NAME, barrier);
+ t.start();
+ threadsA.add(t);
+ }
+ List<CreateLMThread> threadsB = new ArrayList<CreateLMThread>(numThreadsEach);
+ for (int i = 0; i < numThreadsEach; i++) {
+ CreateLMThread t = new CreateLMThread(root0,
+ HierarchicalLedgerManager.NAME, barrier);
+ t.start();
+ threadsB.add(t);
+ }
+
+ barrier.await();
+
+ int numSuccess = 0;
+ int numFails = 0;
+ for (CreateLMThread t : threadsA) {
+ t.join();
+ t.close();
+ if (t.isSuccessful()) {
+ numSuccess++;
+ } else {
+ numFails++;
+ }
+ }
+
+ for (CreateLMThread t : threadsB) {
+ t.join();
+ t.close();
+ if (t.isSuccessful()) {
+ numSuccess++;
+ } else {
+ numFails++;
+ }
+ }
+ assertEquals("Incorrect number of successes", numThreadsEach, numSuccess);
+ assertEquals("Incorrect number of failures", numThreadsEach, numFails);
+ }
+}
\ No newline at end of file