bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject bookkeeper git commit: BOOKKEEPER-867: New Client API to allow applications pass-in EntryId. (Venkateswararao Jujjuri via sijie)
Date Thu, 22 Oct 2015 07:23:29 GMT
Repository: bookkeeper
Updated Branches:
  refs/heads/master 8a3922e00 -> 355ddbc7c


BOOKKEEPER-867: New Client API to allow applications pass-in EntryId. (Venkateswararao Jujjuri
via sijie)


Project: http://git-wip-us.apache.org/repos/asf/bookkeeper/repo
Commit: http://git-wip-us.apache.org/repos/asf/bookkeeper/commit/355ddbc7
Tree: http://git-wip-us.apache.org/repos/asf/bookkeeper/tree/355ddbc7
Diff: http://git-wip-us.apache.org/repos/asf/bookkeeper/diff/355ddbc7

Branch: refs/heads/master
Commit: 355ddbc7c13e39c49a58371d467e87aa80f696d7
Parents: 8a3922e
Author: Sijie Guo <sijie@apache.org>
Authored: Thu Oct 22 00:22:36 2015 -0700
Committer: Sijie Guo <sijie@apache.org>
Committed: Thu Oct 22 00:22:36 2015 -0700

----------------------------------------------------------------------
 .../apache/bookkeeper/client/BKException.java   |  11 +
 .../apache/bookkeeper/client/BookKeeper.java    |  93 +++++
 .../bookkeeper/client/LedgerCreateOp.java       |  15 +-
 .../apache/bookkeeper/client/LedgerHandle.java  | 111 +++++-
 .../bookkeeper/client/LedgerHandleAdv.java      | 214 +++++++++++
 .../apache/bookkeeper/client/PendingAddOp.java  |  17 +
 .../client/BookieWriteLedgerTest.java           | 365 ++++++++++++++++++-
 7 files changed, 815 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BKException.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BKException.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BKException.java
index 3991085..b2355cd 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BKException.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BKException.java
@@ -96,6 +96,8 @@ public abstract class BKException extends Exception {
             return new BKIllegalOpException();
         case Code.AddEntryQuorumTimeoutException:
             return new BKAddEntryQuorumTimeoutException();
+        case Code.DuplicateEntryIdException:
+            return new BKDuplicateEntryIdException();
         default:
             return new BKUnexpectedConditionException();
         }
@@ -128,6 +130,7 @@ public abstract class BKException extends Exception {
         int ClientClosedException = -19;
         int LedgerExistException = -20;
         int AddEntryQuorumTimeoutException = -21;
+        int DuplicateEntryIdException = -22;
 
         int IllegalOpException = -100;
         int LedgerFencedException = -101;
@@ -192,6 +195,8 @@ public abstract class BKException extends Exception {
             return "Bookie protocol version on server is incompatible with client";
         case Code.MetadataVersionException:
             return "Bad ledger metadata version";
+        case Code.DuplicateEntryIdException:
+            return "Attempted to add Duplicate entryId";
         case Code.LedgerFencedException:
             return "Ledger has been fenced off. Some other client must have opened it to
read";
         case Code.UnauthorizedAccessException:
@@ -261,6 +266,12 @@ public abstract class BKException extends Exception {
         }
     }
 
+    public static class BKDuplicateEntryIdException extends BKException {
+        public BKDuplicateEntryIdException() {
+            super(Code.DuplicateEntryIdException);
+        }
+    }
+
     public static class BKUnexpectedConditionException extends BKException {
         public BKUnexpectedConditionException() {
             super(Code.UnexpectedConditionException);

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index 6bb71fa..ed744b0 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -537,6 +537,99 @@ public class BookKeeper {
     }
 
     /**
+     * Synchronous call to create ledger.
+     * Creates a new ledger asynchronously and returns {@link LedgerHandleAdv} which can
accept entryId.
+     * Parameters must match those of
+     * {@link #asyncCreateLedgerAdv(int, int, int, DigestType, byte[],
+     *                           AsyncCallback.CreateCallback, Object)}
+     *
+     * @param ensSize
+     * @param writeQuorumSize
+     * @param ackQuorumSize
+     * @param digestType
+     * @param passwd
+     * @return a handle to the newly created ledger
+     * @throws InterruptedException
+     * @throws BKException
+     */
+    public LedgerHandle createLedgerAdv(int ensSize, int writeQuorumSize, int ackQuorumSize,
+                                        DigestType digestType, byte passwd[])
+            throws InterruptedException, BKException {
+        SyncCounter counter = new SyncCounter();
+        counter.inc();
+        /*
+         * Calls asynchronous version
+         */
+        asyncCreateLedgerAdv(ensSize, writeQuorumSize, ackQuorumSize, digestType, passwd,
+                             new SyncCreateCallback(), counter);
+
+        /*
+         * Wait
+         */
+        counter.block(0);
+        if (counter.getrc() != BKException.Code.OK) {
+            LOG.error("Error while creating ledger : {}", counter.getrc());
+            throw BKException.create(counter.getrc());
+        } else if (counter.getLh() == null) {
+            LOG.error("Unexpected condition : no ledger handle returned for a success ledger
creation");
+            throw BKException.create(BKException.Code.UnexpectedConditionException);
+        }
+
+        return counter.getLh();
+    }
+
+    /**
+     * Creates a new ledger asynchronously and returns {@link LedgerHandleAdv}
+     * which can accept entryId.  Ledgers created with this call have ability to accept
+     * a separate write quorum and ack quorum size. The write quorum must be larger than
+     * the ack quorum.
+     *
+     * Separating the write and the ack quorum allows the BookKeeper client to continue
+     * writing when a bookie has failed but the failure has not yet been detected. Detecting
+     * a bookie has failed can take a number of seconds, as configured by the read timeout
+     * {@link ClientConfiguration#getReadTimeout()}. Once the bookie failure is detected,
+     * that bookie will be removed from the ensemble.
+     *
+     * The other parameters match those of {@link #asyncCreateLedger(int, int, DigestType,
byte[],
+     *                                      AsyncCallback.CreateCallback, Object)}
+     *
+     * @param ensSize
+     *          number of bookies over which to stripe entries
+     * @param writeQuorumSize
+     *          number of bookies each entry will be written to
+     * @param ackQuorumSize
+     *          number of bookies which must acknowledge an entry before the call is completed
+     * @param digestType
+     *          digest type, either MAC or CRC32
+     * @param passwd
+     *          password
+     * @param cb
+     *          createCallback implementation
+     * @param ctx
+     *          optional control object
+     */
+    public void asyncCreateLedgerAdv(final int ensSize,
+                                     final int writeQuorumSize,
+                                     final int ackQuorumSize,
+                                     final DigestType digestType,
+                                     final byte[] passwd, final CreateCallback cb, final
Object ctx) {
+        if (writeQuorumSize < ackQuorumSize) {
+            throw new IllegalArgumentException("Write quorum must be larger than ack quorum");
+        }
+        closeLock.readLock().lock();
+        try {
+            if (closed) {
+                cb.createComplete(BKException.Code.ClientClosedException, null, ctx);
+                return;
+            }
+            new LedgerCreateOp(BookKeeper.this, ensSize, writeQuorumSize,
+                               ackQuorumSize, digestType, passwd, cb, ctx).initiateAdv();
+        } finally {
+            closeLock.readLock().unlock();
+        }
+    }
+
+    /**
      * Open existing ledger asynchronously for reading.
      *
      * Opening a ledger with this method invokes fencing and recovery on the ledger

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
index 7c181b5..6f794d0 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
@@ -54,6 +54,7 @@ class LedgerCreateOp implements GenericCallback<Void> {
     DigestType digestType;
     long startTime;
     OpStatsLogger createOpLogger;
+    boolean adv = false;
 
     /**
      * Constructor
@@ -137,6 +138,14 @@ class LedgerCreateOp implements GenericCallback<Void> {
     }
 
     /**
+     * Initiates the operation to return LedgerHandleAdv.
+     */
+    public void initiateAdv() {
+        this.adv = true;
+        initiate();
+    }
+
+    /**
      * Callback when created ledger.
      */
     @Override
@@ -151,7 +160,11 @@ class LedgerCreateOp implements GenericCallback<Void> {
         }
 
         try {
-            lh = new LedgerHandle(bk, ledgerId, metadata, digestType, passwd);
+            if (adv) {
+                lh = new LedgerHandleAdv(bk, ledgerId, metadata, digestType, passwd);
+            } else {
+                lh = new LedgerHandle(bk, ledgerId, metadata, digestType, passwd);
+            }
         } catch (GeneralSecurityException e) {
             LOG.error("Security exception while creating ledger: " + ledgerId, e);
             createComplete(BKException.Code.DigestNotInitializedException, null);

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
index 36faac4..61cc603 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -78,7 +78,7 @@ public class LedgerHandle {
     final static public long INVALID_ENTRY_ID = BookieProtocol.INVALID_ENTRY_ID;
 
     final AtomicInteger blockAddCompletions = new AtomicInteger(0);
-    final Queue<PendingAddOp> pendingAddOps = new ConcurrentLinkedQueue<PendingAddOp>();
+    Queue<PendingAddOp> pendingAddOps;
 
     final Counter ensembleChangeCounter;
     final Counter lacUpdateHitsCounter;
@@ -89,6 +89,7 @@ public class LedgerHandle {
             throws GeneralSecurityException, NumberFormatException {
         this.bk = bk;
         this.metadata = metadata;
+        this.pendingAddOps = new ConcurrentLinkedQueue<PendingAddOp>();
 
         if (metadata.isClosed()) {
             lastAddConfirmed = lastAddPushed = metadata.getLastEntryId();
@@ -473,6 +474,23 @@ public class LedgerHandle {
     }
 
     /**
+     * Add entry synchronously to an open ledger. This can be used only with
+     * {@link LedgerHandleAdv} returned through ledgers created with {@link
+     * BookKeeper#createLedgerAdv(int, int, int, DigestType, byte[])}.
+     *
+     *
+     * @param entryId
+     *            entryId to be added
+     * @param data
+     *            array of bytes to be written to the ledger
+     * @return the entryId of the new inserted entry
+     */
+    public long addEntry(final long entryId, byte[] data) throws InterruptedException, BKException
{
+        LOG.error("To use this feature Ledger must be created with createLedgerAdv interface.");
+        throw BKException.create(BKException.Code.IllegalOpException);
+    }
+
+    /**
      * Add entry synchronously to an open ledger.
      *
      * @param data
@@ -502,6 +520,27 @@ public class LedgerHandle {
     }
 
     /**
+     * Add entry synchronously to an open ledger. This can be used only with
+     * {@link LedgerHandleAdv} returned through ledgers created with {@link
+     * BookKeeper#createLedgerAdv(int, int, int, DigestType, byte[])}.
+     *
+     * @param entryId
+     *            entryId to be added.
+     * @param data
+     *            array of bytes to be written to the ledger
+     * @param offset
+     *            offset from which to take bytes from data
+     * @param length
+     *            number of bytes to take from data
+     * @return entryId
+     */
+    public long addEntry(final long entryId, byte[] data, int offset, int length) throws
InterruptedException,
+            BKException {
+        LOG.error("To use this feature Ledger must be created with createLedgerAdv() interface.");
+        throw BKException.create(BKException.Code.IllegalOpException);
+    }
+
+    /**
      * Add entry asynchronously to an open ledger.
      *
      * @param data
@@ -517,6 +556,26 @@ public class LedgerHandle {
     }
 
     /**
+     * Add entry asynchronously to an open ledger. This can be used only with
+     * {@link LedgerHandleAdv} returned through ledgers created with {@link
+     * BookKeeper#createLedgerAdv(int, int, int, DigestType, byte[])}.
+     *
+     * @param entryId
+     *            entryId to be added
+     * @param data
+     *            array of bytes to be written
+     * @param cb
+     *            object implementing callbackinterface
+     * @param ctx
+     *            some control object
+     */
+    public void asyncAddEntry(final long entryId, final byte[] data, final AddCallback cb,
final Object ctx)
+            throws BKException {
+        LOG.error("To use this feature Ledger must be created with createLedgerAdv() interface.");
+        cb.addComplete(BKException.Code.IllegalOpException, LedgerHandle.this, entryId, ctx);
+    }
+
+    /**
      * Add entry asynchronously to an open ledger, using an offset and range.
      *
      * @param data
@@ -539,8 +598,35 @@ public class LedgerHandle {
     }
 
     /**
-     * Make a recovery add entry request. Recovery adds can add to a ledger even if
-     * it has been fenced.
+     * Add entry asynchronously to an open ledger, using an offset and range.
+     * This can be used only with {@link LedgerHandleAdv} returned through
+     * ledgers created with {@link BookKeeper#createLedgerAdv(int, int, int, DigestType,
byte[])}.
+     *
+     * @param entryId
+     *            entryId of the entry to add.
+     * @param data
+     *            array of bytes to be written
+     * @param offset
+     *            offset from which to take bytes from data
+     * @param length
+     *            number of bytes to take from data
+     * @param cb
+     *            object implementing callbackinterface
+     * @param ctx
+     *            some control object
+     * @throws ArrayIndexOutOfBoundsException
+     *             if offset or length is negative or offset and length sum to a
+     *             value higher than the length of data.
+     */
+    public void asyncAddEntry(final long entryId, final byte[] data, final int offset, final
int length,
+            final AddCallback cb, final Object ctx) throws BKException {
+        LOG.error("To use this feature Ledger must be created with createLedgerAdv() interface.");
+        cb.addComplete(BKException.Code.IllegalOpException, LedgerHandle.this, entryId, ctx);
+    }
+
+    /**
+     * Make a recovery add entry request. Recovery adds can add to a ledger even
+     * if it has been fenced.
      *
      * This is only valid for bookie and ledger recovery, which may need to replicate
      * entries to a quorum of bookies to ensure data safety.
@@ -553,13 +639,14 @@ public class LedgerHandle {
         doAsyncAddEntry(op, data, offset, length, cb, ctx);
     }
 
-    private void doAsyncAddEntry(final PendingAddOp op, final byte[] data, final int offset,
final int length,
-                                 final AddCallback cb, final Object ctx) {
+    void doAsyncAddEntry(final PendingAddOp op, final byte[] data, final int offset, final
int length,
+                         final AddCallback cb, final Object ctx) {
+
         if (offset < 0 || length < 0
                 || (offset + length) > data.length) {
             throw new ArrayIndexOutOfBoundsException(
-                "Invalid values for offset("+offset
-                +") or length("+length+")");
+                    "Invalid values for offset(" +offset
+                    +") or length("+length+")");
         }
         throttler.acquire();
 
@@ -592,6 +679,7 @@ public class LedgerHandle {
                         cb.addComplete(BKException.Code.LedgerClosedException,
                                 LedgerHandle.this, INVALID_ENTRY_ID, ctx);
                     }
+
                     @Override
                     public String toString() {
                         return String.format("AsyncAddEntryToClosedLedger(lid=%d)", ledgerId);
@@ -701,6 +789,7 @@ public class LedgerHandle {
         }
         ReadLastConfirmedOp.LastConfirmedDataCallback innercb = new ReadLastConfirmedOp.LastConfirmedDataCallback()
{
             AtomicBoolean completed = new AtomicBoolean(false);
+
             @Override
             public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData
data) {
                 if (rc == BKException.Code.OK) {
@@ -842,11 +931,17 @@ public class LedgerHandle {
         // Start from the head of the queue and proceed while there are
         // entries that have had all their responses come back
         PendingAddOp pendingAddOp;
+
         while ((pendingAddOp = pendingAddOps.peek()) != null
                && blockAddCompletions.get() == 0) {
             if (!pendingAddOp.completed) {
                 return;
             }
+            // Check if it is the next entry in the sequence.
+            if (pendingAddOp.entryId != 0 && pendingAddOp.entryId != lastAddConfirmed
+ 1) {
+                LOG.debug("Head of the queue entryId: {} is not lac: {} + 1", pendingAddOp.entryId,
lastAddConfirmed);
+                return;
+            }
             pendingAddOps.remove();
             lastAddConfirmed = pendingAddOp.entryId;
             pendingAddOp.submitCallback(BKException.Code.OK);
@@ -1219,7 +1314,7 @@ public class LedgerHandle {
         }
     }
 
-    private static class SyncAddCallback implements AddCallback {
+    static class SyncAddCallback implements AddCallback {
         long entryId = -1;
 
         /**

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
new file mode 100644
index 0000000..00fcfa7
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
@@ -0,0 +1,214 @@
+/*
+ *
+ * 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.client;
+
+import java.io.Serializable;
+import java.security.GeneralSecurityException;
+import java.util.Comparator;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+
+import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.util.SafeRunnable;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Ledger Advanced handle extends {@link LedgerHandle} to provide API to add entries with
+ * user supplied entryIds. Through this interface Ledger Length may not be accurate wile
the
+ * ledger being written.
+ */
+public class LedgerHandleAdv extends LedgerHandle {
+    final static Logger LOG = LoggerFactory.getLogger(LedgerHandleAdv.class);
+
+    static class PendingOpsComparator implements Comparator<PendingAddOp>, Serializable
{
+        public int compare(PendingAddOp o1, PendingAddOp o2) {
+            return Long.compare(o1.entryId, o2.entryId);
+        }
+    }
+
+    LedgerHandleAdv(BookKeeper bk, long ledgerId, LedgerMetadata metadata, DigestType digestType,
byte[] password)
+            throws GeneralSecurityException, NumberFormatException {
+        super(bk, ledgerId, metadata, digestType, password);
+        pendingAddOps = new PriorityBlockingQueue<PendingAddOp>(10, new PendingOpsComparator());
+    }
+
+
+    /**
+     * Add entry synchronously to an open ledger.
+     *
+     * @param entryId
+     *            entryId of the entry to add
+     * @param data
+     *            array of bytes to be written to the ledger
+     * @return
+     *            entryId that is just created.
+     */
+    @Override
+    public long addEntry(final long entryId, byte[] data) throws InterruptedException, BKException
{
+
+        return addEntry(entryId, data, 0, data.length);
+
+    }
+
+    /**
+     * Add entry synchronously to an open ledger.
+     *
+     * @param entryId
+     *            entryId of the entry to add
+     * @param data
+     *            array of bytes to be written to the ledger
+     * @param offset
+     *            offset from which to take bytes from data
+     * @param length
+     *            number of bytes to take from data
+     * @return The entryId of newly inserted entry.
+     */
+    @Override
+    public long addEntry(final long entryId, byte[] data, int offset, int length) throws
InterruptedException,
+            BKException {
+        LOG.debug("Adding entry {}", data);
+
+        SyncCounter counter = new SyncCounter();
+        counter.inc();
+
+        SyncAddCallback callback = new SyncAddCallback();
+        asyncAddEntry(entryId, data, offset, length, callback, counter);
+
+        counter.block(0);
+
+        if (counter.getrc() != BKException.Code.OK) {
+            throw BKException.create(counter.getrc());
+        }
+        return callback.entryId;
+    }
+
+    /**
+     * Add entry asynchronously to an open ledger.
+     *
+     * @param entryId
+     *            entryId of the entry to add
+     * @param data
+     *            array of bytes to be written
+     * @param cb
+     *            object implementing callbackinterface
+     * @param ctx
+     *            some control object
+     */
+    @Override
+    public void asyncAddEntry(long entryId, byte[] data, AddCallback cb, Object ctx) throws
BKException {
+        asyncAddEntry(entryId, data, 0, data.length, cb, ctx);
+    }
+
+    /**
+     * Add entry asynchronously to an open ledger, using an offset and range.
+     *
+     * @param entryId
+     *            entryId of the entry to add
+     * @param data
+     *            array of bytes to be written
+     * @param offset
+     *            offset from which to take bytes from data
+     * @param length
+     *            number of bytes to take from data
+     * @param cb
+     *            object implementing callbackinterface
+     * @param ctx
+     *            some control object
+     * @throws ArrayIndexOutOfBoundsException
+     *             if offset or length is negative or offset and length sum to a
+     *             value higher than the length of data.
+     */
+
+    public void asyncAddEntry(final long entryId, final byte[] data, final int offset, final
int length,
+            final AddCallback cb, final Object ctx) {
+        PendingAddOp op = new PendingAddOp(this, cb, ctx);
+        op.setEntryId(entryId);
+        if ((entryId <= this.lastAddConfirmed) || pendingAddOps.contains(op)) {
+            LOG.error("Trying to re-add duplicate entryid:{}", entryId);
+            cb.addComplete(BKException.Code.DuplicateEntryIdException,
+                    LedgerHandleAdv.this, entryId, ctx);
+            return;
+        }
+        pendingAddOps.add(op);
+
+        doAsyncAddEntry(op, data, offset, length, cb, ctx);
+    }
+
+    /**
+     * Overriding part is mostly around setting entryId.
+     * Though there may be some code duplication, Choose to have the override routine so
the control flow is
+     * unaltered in the base class.
+     */
+    @Override
+    void doAsyncAddEntry(final PendingAddOp op, final byte[] data, final int offset, final
int length,
+            final AddCallback cb, final Object ctx) {
+        if (offset < 0 || length < 0
+                || (offset + length) > data.length) {
+            throw new ArrayIndexOutOfBoundsException(
+                "Invalid values for offset("+offset
+                +") or length("+length+")");
+        }
+        throttler.acquire();
+
+        if (metadata.isClosed()) {
+            // make sure the callback is triggered in main worker pool
+            try {
+                bk.mainWorkerPool.submit(new SafeRunnable() {
+                    @Override
+                    public void safeRun() {
+                        LOG.warn("Attempt to add to closed ledger: {}", ledgerId);
+                        cb.addComplete(BKException.Code.LedgerClosedException,
+                                LedgerHandleAdv.this, op.getEntryId(), ctx);
+                    }
+                    @Override
+                    public String toString() {
+                        return String.format("AsyncAddEntryToClosedLedger(lid=%d)", ledgerId);
+                    }
+                });
+            } catch (RejectedExecutionException e) {
+                cb.addComplete(bk.getReturnRc(BKException.Code.InterruptedException),
+                        LedgerHandleAdv.this, op.getEntryId(), ctx);
+            }
+            return;
+        }
+
+        try {
+            final long currentLength = addToLength(length);
+
+            bk.mainWorkerPool.submit(new SafeRunnable() {
+                @Override
+                public void safeRun() {
+                    ChannelBuffer toSend = macManager.computeDigestAndPackageForSending(
+                                               op.getEntryId(), lastAddConfirmed, currentLength,
data, offset, length);
+                    op.initiate(toSend, length);
+                }
+            });
+        } catch (RejectedExecutionException e) {
+            cb.addComplete(bk.getReturnRc(BKException.Code.InterruptedException),
+                    LedgerHandleAdv.this, op.getEntryId(), ctx);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
index 4034c35..bc487f6 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
@@ -91,6 +91,10 @@ class PendingAddOp implements WriteCallback, TimerTask {
         writeSet = new HashSet<Integer>(lh.distributionSchedule.getWriteSet(entryId));
     }
 
+    long getEntryId() {
+        return this.entryId;
+    }
+
     void sendWriteRequest(int bookieIndex) {
         int flags = isRecoveryAdd ? BookieProtocol.FLAG_RECOVERY_ADD : BookieProtocol.FLAG_NONE;
 
@@ -249,4 +253,17 @@ class PendingAddOp implements WriteCallback, TimerTask {
         return sb.toString();
     }
 
+    @Override
+    public int hashCode() {
+        return (int) entryId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+       if (o instanceof PendingAddOp) {
+           return (this.entryId == ((PendingAddOp)o).entryId);
+       }
+       return (this == o);
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/355ddbc7/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieWriteLedgerTest.java
----------------------------------------------------------------------
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieWriteLedgerTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieWriteLedgerTest.java
index 7b77c48..692c480 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieWriteLedgerTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieWriteLedgerTest.java
@@ -131,6 +131,48 @@ public class BookieWriteLedgerTest extends
     }
 
     /**
+     * Verify the functionality of Advanced Ledger which returns
+     * LedgerHandleAdv. LedgerHandleAdv takes entryId for addEntry, and let
+     * user manage entryId allocation.
+     *
+     * @throws Exception
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdv() throws Exception {
+        // Create a ledger
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        for (int i = 0; i < numEntriesToWrite; i++) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+
+            entries1.add(entry.array());
+            lh.addEntry(i, entry.array());
+        }
+        // Start one more bookies
+        startNewBookie();
+
+        // Shutdown one bookie in the last ensemble and continue writing
+        ArrayList<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsembles().entrySet().iterator().next()
+                .getValue();
+        killBookie(ensemble.get(0));
+
+        int i = numEntriesToWrite;
+        numEntriesToWrite = numEntriesToWrite + 50;
+        for (; i < numEntriesToWrite; i++) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+
+            entries1.add(entry.array());
+            lh.addEntry(i, entry.array());
+        }
+
+        readEntries(lh, entries1);
+        lh.close();
+    }
+
+    /**
      * Verify asynchronous writing when few bookie failures in last ensemble.
      */
     @Test(timeout=60000)
@@ -203,8 +245,327 @@ public class BookieWriteLedgerTest extends
         lh2.close();
     }
 
-    private void readEntries(LedgerHandle lh, ArrayList<byte[]> entries)
-            throws InterruptedException, BKException {
+    /**
+     * Verify Advanced asynchronous writing with entryIds in reverse order
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdvWithAsyncWritesWithBookieFailures() throws Exception {
+        // Create ledgers
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        lh2 = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+
+        LOG.info("Ledger ID-1: " + lh.getId());
+        LOG.info("Ledger ID-2: " + lh2.getId());
+        SyncObj syncObj1 = new SyncObj();
+        SyncObj syncObj2 = new SyncObj();
+        for (int i = numEntriesToWrite - 1; i >= 0; i--) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+            try {
+                entries1.add(0, entry.array());
+                entries2.add(0, entry.array());
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj1);
+            lh2.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj2);
+        }
+        // Start One more bookie and shutdown one from last ensemble before reading
+        startNewBookie();
+        ArrayList<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsembles().entrySet().iterator().next()
+                .getValue();
+        killBookie(ensemble.get(0));
+
+        // Wait for all entries to be acknowledged for the first ledger
+        synchronized (syncObj1) {
+            while (syncObj1.counter < numEntriesToWrite) {
+                syncObj1.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj1.rc);
+        }
+        // Wait for all entries to be acknowledged for the second ledger
+        synchronized (syncObj2) {
+            while (syncObj2.counter < numEntriesToWrite) {
+                syncObj2.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj2.rc);
+        }
+
+        // Reading ledger till the last entry
+        readEntries(lh, entries1);
+        readEntries(lh2, entries2);
+        lh.close();
+        lh2.close();
+    }
+
+    /**
+     * Verify Advanced asynchronous writing with entryIds in pseudo random order with bookie
failures between writes
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdvWithRandomAsyncWritesWithBookieFailuresBetweenWrites()
throws Exception {
+        // Create ledgers
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        lh2 = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+
+        LOG.info("Ledger ID-1: " + lh.getId());
+        LOG.info("Ledger ID-2: " + lh2.getId());
+        SyncObj syncObj1 = new SyncObj();
+        SyncObj syncObj2 = new SyncObj();
+        int batchSize = 5;
+        int i, j;
+
+        // Fill the result buffers first
+        for (i = 0; i < numEntriesToWrite; i++) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+            try {
+                entries1.add(0, entry.array());
+                entries2.add(0, entry.array());
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        for (i = 0; i < batchSize; i++) {
+            for (j = i; j < numEntriesToWrite; j = j + batchSize) {
+                byte[] entry1 = entries1.get(j);
+                byte[] entry2 = entries2.get(j);
+                lh.asyncAddEntry(j, entry1, 0, entry1.length, this, syncObj1);
+                lh2.asyncAddEntry(j, entry2, 0, entry2.length, this, syncObj2);
+                if (j == numEntriesToWrite/2) {
+                    // Start One more bookie and shutdown one from last ensemble at half-way
+                    startNewBookie();
+                    ArrayList<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsembles().entrySet()
+                            .iterator().next().getValue();
+                    killBookie(ensemble.get(0));
+                }
+            }
+        }
+
+        // Wait for all entries to be acknowledged for the first ledger
+        synchronized (syncObj1) {
+            while (syncObj1.counter < numEntriesToWrite) {
+                syncObj1.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj1.rc);
+        }
+        // Wait for all entries to be acknowledged for the second ledger
+        synchronized (syncObj2) {
+            while (syncObj2.counter < numEntriesToWrite) {
+                syncObj2.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj2.rc);
+        }
+
+        // Reading ledger till the last entry
+        readEntries(lh, entries1);
+        readEntries(lh2, entries2);
+        lh.close();
+        lh2.close();
+    }
+
+    /**
+     * Verify Advanced asynchronous writing with entryIds in pseudo random order
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdvWithRandomAsyncWritesWithBookieFailures() throws Exception
{
+        // Create ledgers
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        lh2 = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+
+        LOG.info("Ledger ID-1: " + lh.getId());
+        LOG.info("Ledger ID-2: " + lh2.getId());
+        SyncObj syncObj1 = new SyncObj();
+        SyncObj syncObj2 = new SyncObj();
+        int batchSize = 5;
+        int i, j;
+
+        // Fill the result buffers first
+        for (i = 0; i < numEntriesToWrite; i++) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+            try {
+                entries1.add(0, entry.array());
+                entries2.add(0, entry.array());
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+        for (i = 0; i < batchSize; i++) {
+            for (j = i; j < numEntriesToWrite; j = j + batchSize) {
+                byte[] entry1 = entries1.get(j);
+                byte[] entry2 = entries2.get(j);
+                lh.asyncAddEntry(j, entry1, 0, entry1.length, this, syncObj1);
+                lh2.asyncAddEntry(j, entry2, 0, entry2.length, this, syncObj2);
+            }
+        }
+        // Start One more bookie and shutdown one from last ensemble before reading
+        startNewBookie();
+        ArrayList<BookieSocketAddress> ensemble = lh.getLedgerMetadata().getEnsembles().entrySet().iterator().next()
+                .getValue();
+        killBookie(ensemble.get(0));
+
+        // Wait for all entries to be acknowledged for the first ledger
+        synchronized (syncObj1) {
+            while (syncObj1.counter < numEntriesToWrite) {
+                syncObj1.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj1.rc);
+        }
+        // Wait for all entries to be acknowledged for the second ledger
+        synchronized (syncObj2) {
+            while (syncObj2.counter < numEntriesToWrite) {
+                syncObj2.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj2.rc);
+        }
+
+        // Reading ledger till the last entry
+        readEntries(lh, entries1);
+        readEntries(lh2, entries2);
+        lh.close();
+        lh2.close();
+    }
+
+    /**
+     * Skips few entries before closing the ledger and assert that the
+     * lastAddConfirmed is right before our skipEntryId.
+     *
+     * @throws Exception
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdvWithSkipEntries() throws Exception {
+        long ledgerId;
+        SyncObj syncObj1 = new SyncObj();
+
+        // Create a ledger
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        // Save ledgerId to reopen the ledger
+        ledgerId = lh.getId();
+        LOG.info("Ledger ID: " + ledgerId);
+        int skipEntryId = rng.nextInt(numEntriesToWrite - 1);
+        for (int i = numEntriesToWrite - 1; i >= 0; i--) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+            try {
+                entries1.add(0, entry.array());
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            if (i == skipEntryId) {
+                LOG.info("Skipping entry:{}", skipEntryId);
+                continue;
+            }
+            lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj1);
+        }
+        // wait for all entries to be acknowledged for the first ledger
+        synchronized (syncObj1) {
+            while (syncObj1.counter < skipEntryId) {
+                syncObj1.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj1.rc);
+        }
+        // Close the ledger
+        lh.close();
+        // Open the ledger
+        lh = bkc.openLedger(ledgerId, digestType, ledgerPassword);
+        assertEquals(lh.lastAddConfirmed, skipEntryId - 1);
+        lh.close();
+    }
+
+    /**
+     * Verify the functionality LedgerHandleAdv addEntry with duplicate entryIds
+     *
+     * @throws Exception
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdvSyncAddDuplicateEntryIds() throws Exception {
+        // Create a ledger
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        LOG.info("Ledger ID: " + lh.getId());
+        for (int i = 0; i < numEntriesToWrite; i++) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+
+            entries1.add(entry.array());
+            lh.addEntry(i, entry.array());
+            entry.position(0);
+        }
+        readEntries(lh, entries1);
+
+        int dupEntryId = rng.nextInt(numEntriesToWrite - 1);
+
+        try {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+            lh.addEntry(dupEntryId, entry.array());
+            fail("Expected exception not thrown");
+        } catch (BKException e) {
+            // This test expects DuplicateEntryIdException
+            assertEquals(e.getCode(), BKException.Code.DuplicateEntryIdException);
+        }
+        lh.close();
+    }
+
+    /**
+     * Verify the functionality LedgerHandleAdv asyncAddEntry with duplicate
+     * entryIds
+     *
+     * @throws Exception
+     */
+    @Test(timeout = 60000)
+    public void testLedgerCreateAdvSyncAsyncAddDuplicateEntryIds() throws Exception {
+        long ledgerId;
+        SyncObj syncObj1 = new SyncObj();
+        SyncObj syncObj2 = new SyncObj();
+
+        // Create a ledger
+        lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword);
+        // Save ledgerId to reopen the ledger
+        ledgerId = lh.getId();
+        LOG.info("Ledger ID: " + ledgerId);
+        for (int i = numEntriesToWrite - 1; i >= 0; i--) {
+            ByteBuffer entry = ByteBuffer.allocate(4);
+            entry.putInt(rng.nextInt(maxInt));
+            entry.position(0);
+            try {
+                entries1.add(0, entry.array());
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj1);
+            if (rng.nextBoolean()) {
+                // Attempt to write the same entry
+                lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj2);
+                synchronized (syncObj2) {
+                    while (syncObj2.counter < 1) {
+                        syncObj2.wait();
+                    }
+                    assertEquals(BKException.Code.DuplicateEntryIdException, syncObj2.rc);
+                }
+            }
+        }
+        // Wait for all entries to be acknowledged for the first ledger
+        synchronized (syncObj1) {
+            while (syncObj1.counter < numEntriesToWrite) {
+                syncObj1.wait();
+            }
+            assertEquals(BKException.Code.OK, syncObj1.rc);
+        }
+        // Close the ledger
+        lh.close();
+    }
+
+    private void readEntries(LedgerHandle lh, ArrayList<byte[]> entries) throws InterruptedException,
BKException {
         ls = lh.readEntries(0, numEntriesToWrite - 1);
         int index = 0;
         while (ls.hasMoreElements()) {


Mime
View raw message