bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mme...@apache.org
Subject [1/2] bookkeeper git commit: BOOKKEEPER-552: 64 Bits Ledger ID Generation
Date Tue, 02 May 2017 21:05:07 GMT
Repository: bookkeeper
Updated Branches:
  refs/heads/master 9c79e078b -> 057af8dbc


http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LongZkLedgerIdGenerator.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LongZkLedgerIdGenerator.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LongZkLedgerIdGenerator.java
new file mode 100644
index 0000000..393f9b1
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LongZkLedgerIdGenerator.java
@@ -0,0 +1,333 @@
+/**
+ * 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.ArrayList;
+import java.util.Formatter;
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
+
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
+import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.AsyncCallback.StringCallback;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ZooKeeper based ledger id generator class, which using EPHEMERAL_SEQUENTIAL
+ * with <i>(ledgerIdGenPath)/HOB-[high-32-bits]/ID-</i> prefix to generate ledger
id. Note
+ * zookeeper sequential counter has a format of %10d -- that is 10 digits with 0
+ * (zero) padding, i.e. "&lt;path&gt;0000000001", so ledger id space would be
+ * fundamentally limited to 9 billion. In practice, the id generated by zookeeper
+ * is only 31 bits (signed 32-bit integer), so the limit is much lower than 9 billion.
+ *
+ * In order to support the full range of the long ledgerId, once ledgerIds reach Integer.MAX_INT,
+ * a new system is employed. The 32 most significant bits of the ledger ID are taken and
turned into
+ * a directory prefixed with <i>HOB-</i> under <i>(ledgerIdGenPath)</i>
+ *
+ * Under this <i>HOB-</i> directory, zookeeper is used to continue generating
EPHEMERAL_SEQUENTIAL ids
+ * which constitute the lower 32-bits of the ledgerId (sign bit is always 0). Once the <i>HOB-</i>
directory runs out of available
+ * ids, the process is repeated. The higher bits are incremented, a new <i>HOB-</i>
directory is created, and
+ * zookeeper generates sequential ids underneath it.
+ *
+ * The reason for treating ids which are less than Integer.MAX_INT differently is to maintain
backwards
+ * compatibility. This is a drop-in replacement for ZkLedgerIdGenerator.
+ */
+public class LongZkLedgerIdGenerator implements LedgerIdGenerator {
+    private static final Logger LOG = LoggerFactory.getLogger(LongZkLedgerIdGenerator.class);
+    private ZooKeeper zk;
+    private String ledgerIdGenPath;
+    private ZkLedgerIdGenerator shortIdGen;
+    private List<String> highOrderDirectories;
+    private HighOrderLedgerIdGenPathStatus ledgerIdGenPathStatus;
+    private final List<ACL> zkAcls;
+
+    private enum HighOrderLedgerIdGenPathStatus {
+        UNKNOWN,
+        PRESENT,
+        NOT_PRESENT
+    };
+
+    public LongZkLedgerIdGenerator(ZooKeeper zk, String ledgersPath, String idGenZnodeName,
ZkLedgerIdGenerator shortIdGen, List<ACL> zkAcls) {
+        this.zk = zk;
+        if (StringUtils.isBlank(idGenZnodeName)) {
+            this.ledgerIdGenPath = ledgersPath;
+        } else {
+            this.ledgerIdGenPath = ledgersPath + "/" + idGenZnodeName;
+        }
+        this.shortIdGen = shortIdGen;
+        highOrderDirectories = new ArrayList<String>();
+        ledgerIdGenPathStatus = HighOrderLedgerIdGenPathStatus.UNKNOWN;
+        this.zkAcls = zkAcls; 
+    }
+
+    private void generateLongLedgerIdLowBits(final String ledgerPrefix, long highBits, final
GenericCallback<Long> cb) throws KeeperException, InterruptedException, IOException
{
+        String highPath = ledgerPrefix + formatHalfId((int)highBits);
+        ZkLedgerIdGenerator.generateLedgerIdImpl(new GenericCallback<Long>(){
+            @Override
+            public void operationComplete(int rc, Long result) {
+                if(rc == BKException.Code.OK) {
+                    assert((highBits & 0xFFFFFFFF00000000l) == 0);
+                    assert((result & 0xFFFFFFFF00000000l) == 0);
+                    cb.operationComplete(rc, (highBits << 32) | result);
+                }
+                else if(rc == BKException.Code.LedgerIdOverflowException) {
+                    // Lower bits are full. Need to expand and create another HOB node.
+                    try {
+                        Long newHighBits = highBits + 1;
+                        createHOBPathAndGenerateId(ledgerPrefix, newHighBits.intValue(),
cb);
+                    }
+                    catch (KeeperException e) {
+                        LOG.error("Failed to create long ledger ID path", e);
+                        cb.operationComplete(BKException.Code.ZKException, null);
+                    }
+                    catch (InterruptedException e) {
+                        LOG.error("Failed to create long ledger ID path", e);
+                        cb.operationComplete(BKException.Code.InterruptedException, null);
+                    } catch (IOException e) {
+                        LOG.error("Failed to create long ledger ID path", e);
+                        cb.operationComplete(BKException.Code.IllegalOpException, null);
+                    }
+
+                }
+            }
+
+        }, zk, ZkLedgerIdGenerator.createLedgerPrefix(highPath, null), zkAcls);
+    }
+
+    /**
+     * Formats half an ID as 10-character 0-padded string
+     * @param i - 32 bits of the ID to format
+     * @return a 10-character 0-padded string.
+     */
+    private String formatHalfId(int i) {
+        return String.format("%010d", i);
+    }
+
+    private void createHOBPathAndGenerateId(String ledgerPrefix, int hob, final GenericCallback<Long>
cb) throws KeeperException, InterruptedException, IOException {
+        try {
+            LOG.debug("Creating HOB path: {}", ledgerPrefix + formatHalfId(hob));
+            zk.create(ledgerPrefix + formatHalfId(hob), new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+        }
+        catch(KeeperException.NodeExistsException e) {
+            // It's fine if we lost a race to create the node (NodeExistsException).
+            // All other exceptions should continue unwinding.
+            LOG.debug("Tried to create High-order-bits node, but it already existed!", e);
+        }
+        // We just created a new HOB directory. Invalidate the directory cache
+        invalidateDirectoryCache();
+        generateLongLedgerId(cb); // Try again.
+    }
+
+    private void invalidateDirectoryCache() {
+        highOrderDirectories = null;
+    }
+
+    private void generateLongLedgerId(final GenericCallback<Long> cb) throws KeeperException,
InterruptedException, IOException {
+        final String hobPrefix = "HOB-";
+        final String ledgerPrefix = this.ledgerIdGenPath + "/" + hobPrefix;
+
+        // Only pull the directories from zk if we don't have any cached.
+        boolean refreshedDirectories = false;
+        if(highOrderDirectories == null) {
+            refreshedDirectories = true;
+            highOrderDirectories = zk.getChildren(ledgerIdGenPath, false);
+        }
+
+        Optional<Long> largest = highOrderDirectories.stream()
+            .map((t) -> {
+                    try {
+                        return Long.parseLong(t.replace(hobPrefix, ""));
+                    }
+                    catch(NumberFormatException e) {
+                        return null;
+                    }
+                })
+            .filter((t) -> t != null)
+            .reduce(Math::max);
+
+        // If we didn't get any valid IDs from the directory...
+        if(!largest.isPresent()) {
+            if(!refreshedDirectories) {
+                // Our cache might be bad. Invalidate it and retry.
+                invalidateDirectoryCache();
+                generateLongLedgerId(cb); // Try again
+            }
+            else {
+                // else, Start at HOB-0000000001;
+                createHOBPathAndGenerateId(ledgerPrefix, 1, cb);
+            }
+            return;
+        }
+
+        // Found the largest.
+        // Get the low-order bits.
+        final Long highBits = largest.get();
+        generateLongLedgerIdLowBits(ledgerPrefix, highBits, cb);
+
+        // Perform garbage collection on HOB- directories.
+        // Keeping 3 should be plenty to prevent races
+        if(highOrderDirectories.size() > 3) {
+            Object[] highOrderDirs = highOrderDirectories.stream()
+                    .map((t) -> {
+                            try {
+                                return Long.parseLong(t.replace(hobPrefix, ""));
+                            }
+                            catch(NumberFormatException e) {
+                                return null;
+                            }
+                        })
+                    .filter((t) -> t != null)
+                    .sorted()
+                    .toArray();
+
+            // Go ahead and invalidate. We want to reload cache even if we fail.
+            invalidateDirectoryCache();
+
+            for(int i = 0; i < highOrderDirs.length - 3; i++) {
+                String path = ledgerPrefix + formatHalfId(((Long)highOrderDirs[i]).intValue());
+                LOG.debug("DELETING HIGH ORDER DIR: {}", path);
+                try {
+                    zk.delete(path, 0);
+                }
+                catch (KeeperException e) {
+                    // We don't care if we fail. Just warn about it.
+                    LOG.debug("Failed to delete {}", path);
+                }
+            }
+        }
+    }
+
+    private void createLongLedgerIdPathAndGenerateLongLedgerId(final GenericCallback<Long>
cb, String createPath) {
+        ZkUtils.asyncCreateFullPathOptimistic(zk, ledgerIdGenPath, new byte[0], Ids.OPEN_ACL_UNSAFE,
+                                              CreateMode.PERSISTENT, new StringCallback()
{
+                                                      @Override
+                                                      public void processResult(int rc, String
path, Object ctx, String name) {
+                                                          try {
+                                                              setLedgerIdGenPathStatus(HighOrderLedgerIdGenPathStatus.PRESENT);
+                                                              generateLongLedgerId(cb);
+                                                          } catch (KeeperException e) {
+                                                              LOG.error("Failed to create
long ledger ID path", e);
+                                                              setLedgerIdGenPathStatus(HighOrderLedgerIdGenPathStatus.UNKNOWN);
+                                                              cb.operationComplete(BKException.Code.ZKException,
null);
+                                                          } catch (InterruptedException e)
{
+                                                              LOG.error("Failed to create
long ledger ID path", e);
+                                                              setLedgerIdGenPathStatus(HighOrderLedgerIdGenPathStatus.UNKNOWN);
+                                                              cb.operationComplete(BKException.Code.InterruptedException,
null);
+                                                          } catch (IOException e) {
+                                                              LOG.error("Failed to create
long ledger ID path", e);
+                                                              setLedgerIdGenPathStatus(HighOrderLedgerIdGenPathStatus.UNKNOWN);
+                                                              cb.operationComplete(BKException.Code.IllegalOpException,
null);
+                                                          }
+                                                      }
+                                                  }, null);
+    }
+
+    public void invalidateLedgerIdGenPathStatus() {
+        setLedgerIdGenPathStatus(HighOrderLedgerIdGenPathStatus.UNKNOWN);
+    }
+
+    synchronized private void setLedgerIdGenPathStatus(HighOrderLedgerIdGenPathStatus status)
{
+        ledgerIdGenPathStatus = status;
+    }
+
+    /**
+     * Checks the existence of the long ledger id gen path. Existence indicates we have switched
from the legacy
+     * algorithm to the new method of generating 63-bit ids. If the existence is UNKNOWN,
it looks in zk to
+     * find out. If it previously checked in zk, it returns that value. This value changes
when we run out
+     * of ids < Integer.MAX_VALUE, and try to create the long ledger id gen path.
+     * @see createLongLedgerIdPathAndGenerateLongLedgerId
+     * @param zk
+     * @return Does the long ledger id gen path exist?
+     * @throws KeeperException
+     * @throws InterruptedException
+     */
+    synchronized public boolean ledgerIdGenPathPresent(ZooKeeper zk) throws KeeperException,
InterruptedException {
+        switch(ledgerIdGenPathStatus) {
+        case UNKNOWN:
+            if(zk.exists(ledgerIdGenPath, false) != null) {
+                ledgerIdGenPathStatus = HighOrderLedgerIdGenPathStatus.PRESENT;
+                return true;
+            }
+            else {
+                ledgerIdGenPathStatus = HighOrderLedgerIdGenPathStatus.NOT_PRESENT;
+                return false;
+            }
+        case PRESENT:
+            return true;
+        case NOT_PRESENT:
+            return false;
+        default:
+            return false;
+        }
+    }
+
+    @Override
+    public void generateLedgerId(final GenericCallback<Long> cb) {
+        try {
+            if(!ledgerIdGenPathPresent(zk)) {
+                // We've not moved onto 63-bit ledgers yet.
+                shortIdGen.generateLedgerId(new GenericCallback<Long>(){
+                        @Override
+                        public void operationComplete(int rc, Long result) {
+                            if(rc == BKException.Code.LedgerIdOverflowException) {
+                                // 31-bit IDs overflowed. Start using 63-bit ids.
+                                createLongLedgerIdPathAndGenerateLongLedgerId(cb, ledgerIdGenPath);
+                            }
+                            else {
+                                // 31-bit Generation worked OK, or had some other
+                                // error that we will pass on.
+                                cb.operationComplete(rc, result);
+                            }
+                        }
+                    });
+            }
+            else {
+                // We've already started generating 63-bit ledger IDs.
+                // Keep doing that.
+                generateLongLedgerId(cb);
+            }
+        } catch (KeeperException e) {
+            LOG.error("Failed to create long ledger ID path", e);
+            cb.operationComplete(BKException.Code.ZKException, null);
+        }
+        catch (InterruptedException e) {
+            LOG.error("Failed to create long ledger ID path", e);
+            cb.operationComplete(BKException.Code.InterruptedException, null);
+        }
+        catch (IOException e) {
+            LOG.error("Failed to create long ledger ID path", e);
+            cb.operationComplete(BKException.Code.IllegalOpException, null);
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        shortIdGen.close();
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerIdGenerator.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerIdGenerator.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerIdGenerator.java
index a2e79af..1373f34 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerIdGenerator.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerIdGenerator.java
@@ -47,7 +47,6 @@ public class ZkLedgerIdGenerator implements LedgerIdGenerator {
     static final String LEDGER_ID_GEN_PREFIX = "ID-";
 
     final ZooKeeper zk;
-    final String ledgerIdGenPath;
     final String ledgerPrefix;
     final List<ACL> zkAcls;
 
@@ -56,17 +55,26 @@ public class ZkLedgerIdGenerator implements LedgerIdGenerator {
                                String idGenZnodeName,
                                List<ACL> zkAcls) {
         this.zk = zk;
+        ledgerPrefix = createLedgerPrefix(ledgersPath, idGenZnodeName);
         this.zkAcls = zkAcls;
+    }
+
+    public static String createLedgerPrefix(String ledgersPath, String idGenZnodeName) {
+        String ledgerIdGenPath = null;
         if (StringUtils.isBlank(idGenZnodeName)) {
-            this.ledgerIdGenPath = ledgersPath;
+            ledgerIdGenPath = ledgersPath;
         } else {
-            this.ledgerIdGenPath = ledgersPath + "/" + idGenZnodeName;
+            ledgerIdGenPath = ledgersPath + "/" + idGenZnodeName;
         }
-        this.ledgerPrefix = this.ledgerIdGenPath + "/" + LEDGER_ID_GEN_PREFIX;
+        return ledgerIdGenPath + "/" + LEDGER_ID_GEN_PREFIX;
     }
 
     @Override
     public void generateLedgerId(final GenericCallback<Long> cb) {
+        generateLedgerIdImpl(cb, zk, ledgerPrefix, zkAcls);
+    }
+
+    public static void generateLedgerIdImpl(final GenericCallback<Long> cb, ZooKeeper
zk, String ledgerPrefix, List<ACL> zkAcls) {
         ZkUtils.asyncCreateFullPathOptimistic(zk, ledgerPrefix, new byte[0], zkAcls,
                 CreateMode.EPHEMERAL_SEQUENTIAL,
                 new StringCallback() {
@@ -84,8 +92,13 @@ public class ZkLedgerIdGenerator implements LedgerIdGenerator {
                          */
                         long ledgerId;
                         try {
-                            ledgerId = getLedgerIdFromGenPath(idPathName);
-                            cb.operationComplete(BKException.Code.OK, ledgerId);
+                            ledgerId = getLedgerIdFromGenPath(idPathName, ledgerPrefix);
+                            if(ledgerId < 0 || ledgerId >= Integer.MAX_VALUE) {
+                                cb.operationComplete(BKException.Code.LedgerIdOverflowException,
null);
+                            }
+                            else {
+                                cb.operationComplete(BKException.Code.OK, ledgerId);
+                            }
                         } catch (IOException e) {
                             LOG.error("Could not extract ledger-id from id gen path:" + path,
e);
                             cb.operationComplete(BKException.Code.ZKException, null);
@@ -109,7 +122,7 @@ public class ZkLedgerIdGenerator implements LedgerIdGenerator {
     }
 
     // get ledger id from generation path
-    private long getLedgerIdFromGenPath(String nodeName) throws IOException {
+    private static long getLedgerIdFromGenPath(String nodeName, String ledgerPrefix) throws
IOException {
         long ledgerId;
         try {
             String parts[] = nodeName.split(ledgerPrefix);

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java
index c2f658b..c0c110d 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/StringUtils.java
@@ -57,7 +57,7 @@ public class StringUtils {
      *          ledger id
      * @return the hierarchical path
      */
-    public static String getHierarchicalLedgerPath(long ledgerId) {
+    public static String getShortHierarchicalLedgerPath(long ledgerId) {
         String ledgerIdStr = getZKStringId(ledgerId);
         // do 2-4-4 split
         StringBuilder sb = new StringBuilder();
@@ -90,6 +90,13 @@ public class StringUtils {
         return sb.toString();
     }
     
+    public static String getHybridHierarchicalLedgerPath(long ledgerId) {
+        if(ledgerId < Integer.MAX_VALUE) {
+            return getShortHierarchicalLedgerPath(ledgerId);
+        }
+        return getLongHierarchicalLedgerPath(ledgerId);
+    }
+    
     /**
      * Parse the hierarchical ledger path to its ledger id
      *
@@ -119,7 +126,7 @@ public class StringUtils {
             throws IOException {
         String[] longHierarchicalParts = longHierarchicalLedgerPath.split("/");
         if (longHierarchicalParts.length != 5) {
-            throw new IOException("it is not a valid hierarchical path name : " + longHierarchicalLedgerPath);
+            return stringToHierarchicalLedgerId(longHierarchicalLedgerPath);
         }
         longHierarchicalParts[4] =
                 longHierarchicalParts[4].substring(LEDGER_NODE_PREFIX.length());

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
index 777b619..d25bd70 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
@@ -24,6 +24,7 @@ import org.apache.bookkeeper.client.AsyncCallback.RecoverCallback;
 import org.apache.bookkeeper.client.BookKeeper.DigestType;
 import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory;
 import org.apache.bookkeeper.meta.MSLedgerManagerFactory;
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
@@ -882,10 +883,13 @@ public class BookieRecoveryTest extends MultiLedgerManagerMultiDigestTestCase
{
     public void ensurePasswordUsedForOldLedgers() throws Exception {
         // This test bases on creating old ledgers in version 4.1.0, which only
         // supports ZooKeeper based flat and hierarchical LedgerManagerFactory.
-        // So we ignore it for MSLedgerManagerFactory.
+        // So we ignore it for MSLedgerManagerFactory and LongHierarchicalLedgerManagerFactory.
         if (MSLedgerManagerFactory.class.getName().equals(ledgerManagerFactory)) {
             return;
         }
+        if (LongHierarchicalLedgerManagerFactory.class.getName().equals(ledgerManagerFactory))
{
+            return;
+        }
 
         // stop all bookies
         // and wipe the ledger layout so we can use an old client

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLongZkLedgerIdGenerator.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLongZkLedgerIdGenerator.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLongZkLedgerIdGenerator.java
new file mode 100644
index 0000000..75422a4
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLongZkLedgerIdGenerator.java
@@ -0,0 +1,145 @@
+/**
+ * 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.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
+import org.apache.bookkeeper.test.ZooKeeperUtil;
+import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.ZooDefs.Ids;
+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 TestLongZkLedgerIdGenerator extends TestCase {
+    private static final Logger LOG = LoggerFactory.getLogger(TestZkLedgerIdGenerator.class);
+
+    ZooKeeperUtil zkutil;
+    ZooKeeper zk;
+
+    LongZkLedgerIdGenerator ledgerIdGenerator;
+
+    @Override
+    @Before
+    public void setUp() throws Exception {
+        LOG.info("Setting up test");
+        super.setUp();
+
+        zkutil = new ZooKeeperUtil();
+        zkutil.startServer();
+        zk = zkutil.getZooKeeperClient();
+
+        ZkLedgerIdGenerator shortLedgerIdGenerator = new ZkLedgerIdGenerator(zk,
+                "/test-zk-ledger-id-generator", "idgen", ZooDefs.Ids.OPEN_ACL_UNSAFE);
+        ledgerIdGenerator = new LongZkLedgerIdGenerator(zk, 
+                "/test-zk-ledger-id-generator", "idgen-long", shortLedgerIdGenerator, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+    }
+
+    @Override
+    @After
+    public void tearDown() throws Exception {
+        LOG.info("Tearing down test");
+        ledgerIdGenerator.close();
+        zk.close();
+        zkutil.killServer();
+
+        super.tearDown();
+    }
+
+    @Test(timeout=60000)
+    public void testGenerateLedgerId() throws Exception {
+        // Create *nThread* threads each generate *nLedgers* ledger id,
+        // and then check there is no identical ledger id.
+        final int nThread = 2;
+        final int nLedgers = 2000;
+        // Multiply by two. We're going to do half in the old legacy space and half in the
new.
+        final CountDownLatch countDownLatch = new CountDownLatch(nThread*nLedgers*2); 
+
+        final AtomicInteger errCount = new AtomicInteger(0);
+        final ConcurrentLinkedQueue<Long> ledgerIds = new ConcurrentLinkedQueue<Long>();
+        final GenericCallback<Long> cb = new GenericCallback<Long>() {
+            @Override
+            public void operationComplete(int rc, Long result) {
+                if (Code.OK.intValue() == rc) {
+                    ledgerIds.add(result);
+                } else {
+                    errCount.incrementAndGet();
+                }
+                countDownLatch.countDown();
+            }
+        };
+
+        long start = System.currentTimeMillis();
+
+        for (int i = 0; i < nThread; i++) {
+            new Thread() {
+                @Override
+                public void run() {
+                    for (int j = 0; j < nLedgers; j++) {
+                        ledgerIdGenerator.generateLedgerId(cb);
+                    }
+                }
+            }.start();
+        }
+        
+        // Go and create the long-id directory in zookeeper. This should cause the id generator
to generate ids with the
+        // new algo once we clear it's stored status.
+        ZkUtils.createFullPathOptimistic(zk, "/test-zk-ledger-id-generator/idgen-long", new
byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        ledgerIdGenerator.invalidateLedgerIdGenPathStatus();
+        
+        for (int i = 0; i < nThread; i++) {
+            new Thread() {
+                @Override
+                public void run() {
+                    for (int j = 0; j < nLedgers; j++) {
+                        ledgerIdGenerator.generateLedgerId(cb);
+                    }
+                }
+            }.start();
+        }
+
+        assertTrue("Wait ledger id generation threads to stop timeout : ",
+                countDownLatch.await(30, TimeUnit.SECONDS));
+        LOG.info("Number of generated ledger id: {}, time used: {}", ledgerIds.size(),
+                System.currentTimeMillis() - start);
+        assertEquals("Error occur during ledger id generation : ", 0, errCount.get());
+
+        Set<Long> ledgers = new HashSet<Long>();
+        while (!ledgerIds.isEmpty()) {
+            Long ledger = ledgerIds.poll();
+            assertNotNull("Generated ledger id is null : ", ledger);
+            assertFalse("Ledger id [" + ledger + "] conflict : ", ledgers.contains(ledger));
+            ledgers.add(ledger);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerMultiDigestTestCase.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerMultiDigestTestCase.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerMultiDigestTestCase.java
index 2c9a1f4..357bd34 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerMultiDigestTestCase.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerMultiDigestTestCase.java
@@ -48,6 +48,7 @@ public abstract class MultiLedgerManagerMultiDigestTestCase extends BookKeeperCl
     public static Collection<Object[]> configs() {
         String[] ledgerManagers = {
             "org.apache.bookkeeper.meta.FlatLedgerManagerFactory",
+            "org.apache.bookkeeper.meta.LegacyHierarchicalLedgerManagerFactory",
             "org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory",
             "org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory",
             "org.apache.bookkeeper.meta.MSLedgerManagerFactory",

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerTestCase.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerTestCase.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerTestCase.java
index 34a22af..cb640b1 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerTestCase.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/MultiLedgerManagerTestCase.java
@@ -42,6 +42,7 @@ public abstract class MultiLedgerManagerTestCase extends BookKeeperClusterTestCa
     public static Collection<Object[]> configs() {
         String[] ledgerManagers = new String[] {
             "org.apache.bookkeeper.meta.FlatLedgerManagerFactory",
+            "org.apache.bookkeeper.meta.LegacyHierarchicalLedgerManagerFactory",
             "org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory",
             "org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory",
             "org.apache.bookkeeper.meta.MSLedgerManagerFactory",

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/057af8db/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
index 3249194..18504e3 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/TestBackwardCompat.java
@@ -37,6 +37,7 @@ import org.apache.bookkeeper.bookie.Bookie;
 import org.apache.bookkeeper.bookie.BookieException;
 import org.apache.bookkeeper.bookie.FileSystemUpgrade;
 import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.AbstractConfiguration;
 import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.conf.TestBKConfiguration;
 import org.apache.bookkeeper.util.IOUtils;
@@ -437,6 +438,17 @@ public class TestBackwardCompat {
             return new LedgerCurrent(newbk, newlh);
         }
 
+        static LedgerCurrent openLedger(long id, ClientConfiguration conf) throws Exception
{
+            conf.setZkServers(zkUtil.getZooKeeperConnectString());
+            org.apache.bookkeeper.client.BookKeeper newbk
+                = new org.apache.bookkeeper.client.BookKeeper(conf);
+            org.apache.bookkeeper.client.LedgerHandle newlh
+                = newbk.openLedger(id,
+                                   org.apache.bookkeeper.client.BookKeeper.DigestType.CRC32,
+                                "foobar".getBytes());
+            return new LedgerCurrent(newbk, newlh);
+        }
+
         long getId() {
             return lh.getId();
         }
@@ -827,4 +839,45 @@ public class TestBackwardCompat {
         oldledger.close();
         scur.stop();
     }
+
+    /**
+     * Test compatability between version old version and the current version
+     * with respect to the HierarchicalLedgerManagers.
+     * - 4.2.0 server starts with HierarchicalLedgerManager.
+     * - Write ledgers with old and new clients
+     * - Read ledgers written by old clients.
+     */
+    @Test(timeout = 60000)
+    public void testCompatHierarchicalLedgerManager() throws Exception {
+        File journalDir = createTempDir("bookie", "journal");
+        File ledgerDir = createTempDir("bookie", "ledger");
+
+        int port = PortManager.nextFreePort();
+        // start server, upgrade
+        Server420 s420 = new Server420(journalDir, ledgerDir, port);
+        s420.getConf().setLedgerManagerFactoryClassName("org.apache.bk_v4_2_0.bookkeeper.meta.HierarchicalLedgerManagerFactory");
+        s420.start();
+
+        Ledger420 l420 = Ledger420.newLedger();
+        l420.write100();
+        long oldLedgerId = l420.getId();
+        l420.close();
+        s420.stop();
+
+        // Start the current server
+        ServerCurrent scur = new ServerCurrent(journalDir, ledgerDir, port, true);
+        scur.getConf().setLedgerManagerFactoryClassName("org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory");
+        scur.getConf().setProperty(AbstractConfiguration.LEDGER_MANAGER_FACTORY_DISABLE_CLASS_CHECK,
true);
+        scur.start();
+
+        // Munge the conf so we can test.
+        ClientConfiguration conf = new ClientConfiguration();
+        conf.setLedgerManagerFactoryClassName("org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory");
+        conf.setProperty(AbstractConfiguration.LEDGER_MANAGER_FACTORY_DISABLE_CLASS_CHECK,
true);
+
+        // check that new client can read old ledgers on new server
+        LedgerCurrent oldledger = LedgerCurrent.openLedger(oldLedgerId, conf);
+        assertEquals("Failed to read entries!", 100, oldledger.readAll());
+        oldledger.close();
+    }
 }


Mime
View raw message