diff --git a/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java b/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
index ccdc52b..6c6bd4a 100644
--- a/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
+++ b/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
@@ -21,27 +21,27 @@
 import io.netty.buffer.Unpooled;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
-import java.util.Enumeration;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException.BKNoSuchEntryException;
+import org.apache.bookkeeper.client.BKException.Code;
 import org.apache.bookkeeper.client.DistributionSchedule.WriteSet;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
+import org.apache.bookkeeper.common.concurrent.FutureEventListener;
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.proto.BookieClient;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Reader used for DL tools to read entries.
  */
 public class LedgerReader {
 
-    private static final Logger logger = LoggerFactory.getLogger(LedgerReader.class);
-
     /**
      * Read Result Holder.
      */
@@ -137,21 +137,42 @@
                                                     final GenericCallback<List<LedgerEntry>> callback) {
         final List<LedgerEntry> resultList = new ArrayList<LedgerEntry>();
 
-        final AsyncCallback.ReadCallback readCallback = new AsyncCallback.ReadCallback() {
+        final FutureEventListener<LedgerEntries> readListener = new FutureEventListener<LedgerEntries>() {
+
+            private void readNext(long entryId) {
+                PendingReadOp op = new PendingReadOp(lh, lh.bk.scheduler, entryId, entryId, false);
+                op.future().whenComplete(this);
+                op.submit();
+            }
+
             @Override
-            public void readComplete(int rc, LedgerHandle lh, Enumeration<LedgerEntry> entries, Object ctx) {
-                if (BKException.Code.NoSuchEntryException == rc) {
-                    callback.operationComplete(BKException.Code.OK, resultList);
-                } else if (BKException.Code.OK == rc) {
-                    while (entries.hasMoreElements()) {
-                        resultList.add(entries.nextElement());
-                    }
-                    long entryId = (Long) ctx;
-                    ++entryId;
-                    PendingReadOp readOp = new PendingReadOp(lh, lh.bk.scheduler, entryId, entryId, this, entryId);
-                    readOp.initiate();
+            public void onSuccess(LedgerEntries ledgerEntries) {
+                long entryId = -1L;
+                for (org.apache.bookkeeper.client.api.LedgerEntry entry : ledgerEntries) {
+                    resultList.add(new LedgerEntry((LedgerEntryImpl) entry));
+                    entryId = entry.getEntryId();
+                }
+                try {
+                    ledgerEntries.close();
+                } catch (Exception e) {
+                    // bk should not throw any exceptions here
+                }
+                ++entryId;
+                readNext(entryId);
+            }
+
+            @Override
+            public void onFailure(Throwable throwable) {
+                if (throwable instanceof BKNoSuchEntryException) {
+                    callback.operationComplete(Code.OK, resultList);
                 } else {
-                    callback.operationComplete(rc, resultList);
+                    int retCode;
+                    if (throwable instanceof BKException) {
+                        retCode = ((BKException) throwable).getCode();
+                    } else {
+                        retCode = Code.UnexpectedConditionException;
+                    }
+                    callback.operationComplete(retCode, resultList);
                 }
             }
         };
@@ -169,13 +190,9 @@
             }
 
             long entryId = recoveryData.lastAddConfirmed;
-            PendingReadOp readOp = new PendingReadOp(lh, lh.bk.scheduler, entryId, entryId, readCallback, entryId);
-            try {
-                readOp.initiate();
-            } catch (Throwable t) {
-                logger.error("Failed to initialize pending read entry {} for ledger {} : ",
-                             new Object[] { entryId, lh.getLedgerMetadata(), t });
-            }
+            PendingReadOp op = new PendingReadOp(lh, lh.bk.scheduler, entryId, entryId, false);
+            op.future().whenComplete(readListener);
+            op.submit();
         };
         // Read Last AddConfirmed
         new ReadLastConfirmedOp(lh, readLACCallback).initiate();
diff --git a/pom.xml b/pom.xml
index 616d755..738ef81 100644
--- a/pom.xml
+++ b/pom.xml
@@ -100,7 +100,7 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <!-- dependencies -->
-    <bookkeeper.version>4.6.0-SNAPSHOT</bookkeeper.version>
+    <bookkeeper.version>4.7.0-SNAPSHOT</bookkeeper.version>
     <codahale.metrics.version>3.0.1</codahale.metrics.version>
     <commons-cli.version>1.1</commons-cli.version>
     <commons-codec.version>1.6</commons-codec.version>