HDFS-4130. BKJM: The reading for editlog at NN starting using bkjm is not efficient. Contributed by Han Xiao.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1419661 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -148,6 +148,9 @@ Release 2.0.3-alpha - Unreleased
     HDFS-3680. Allow customized audit logging in HDFS FSNamesystem. (Marcelo
     Vanzin via atm)
 
+    HDFS-4130. BKJM: The reading for editlog at NN starting using bkjm is not efficient.
+    (Han Xiao via umamahesh)
+
   OPTIMIZATIONS
 
   BUG FIXES 

hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/BookKeeperJournalManager.java

@@ -500,16 +500,18 @@ public class BookKeeperJournalManager implements JournalManager {
     } 
   }
 
-  EditLogInputStream getInputStream(long fromTxId, boolean inProgressOk)
+  @Override
-      throws IOException {
+  public void selectInputStreams(Collection<EditLogInputStream> streams,
-    for (EditLogLedgerMetadata l : getLedgerList(inProgressOk)) {
+      long fromTxId, boolean inProgressOk) throws IOException {
-      long lastTxId = l.getLastTxId();
+    List<EditLogLedgerMetadata> currentLedgerList = getLedgerList(inProgressOk);
-      if (l.isInProgress()) {
+    try {
-        lastTxId = recoverLastTxId(l, false);
+      BookKeeperEditLogInputStream elis = null;
-      }
+      for (EditLogLedgerMetadata l : currentLedgerList) {
-
+        long lastTxId = l.getLastTxId();
-      if (fromTxId >= l.getFirstTxId() && fromTxId <= lastTxId) {
+        if (l.isInProgress()) {
-        try {
+          lastTxId = recoverLastTxId(l, false);
+        }
+        if (fromTxId >= l.getFirstTxId() && fromTxId <= lastTxId) {
           LedgerHandle h;
           if (l.isInProgress()) { // we don't want to fence the current journal
             h = bkc.openLedgerNoRecovery(l.getLedgerId(),
@@ -518,42 +520,22 @@ public class BookKeeperJournalManager implements JournalManager {
             h = bkc.openLedger(l.getLedgerId(), BookKeeper.DigestType.MAC,
                 digestpw.getBytes());
           }
-          BookKeeperEditLogInputStream s = new BookKeeperEditLogInputStream(h,
+          elis = new BookKeeperEditLogInputStream(h, l);
-              l);
+          elis.skipTo(fromTxId);
-          s.skipTo(fromTxId);
+        } else {
-          return s;
+          return;
-        } catch (BKException e) {
-          throw new IOException("Could not open ledger for " + fromTxId, e);
-        } catch (InterruptedException ie) {
-          Thread.currentThread().interrupt();
-          throw new IOException("Interrupted opening ledger for "
-                                         + fromTxId, ie);
         }
+        streams.add(elis);
+        if (elis.getLastTxId() == HdfsConstants.INVALID_TXID) {
+          return;
+        }
+        fromTxId = elis.getLastTxId() + 1;
       }
-    }
+    } catch (BKException e) {
-    return null;
+      throw new IOException("Could not open ledger for " + fromTxId, e);
-  }
+    } catch (InterruptedException ie) {
-
+      Thread.currentThread().interrupt();
-  @Override
+      throw new IOException("Interrupted opening ledger for " + fromTxId, ie);
-  public void selectInputStreams(Collection<EditLogInputStream> streams,
-      long fromTxId, boolean inProgressOk) {
-    // NOTE: could probably be rewritten more efficiently
-    while (true) {
-      EditLogInputStream elis;
-      try {
-        elis = getInputStream(fromTxId, inProgressOk);
-      } catch (IOException e) {
-        LOG.error(e);
-        return;
-      }
-      if (elis == null) {
-        return;
-      }
-      streams.add(elis);
-      if (elis.getLastTxId() == HdfsConstants.INVALID_TXID) {
-        return;
-      }
-      fromTxId = elis.getLastTxId() + 1;
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/test/java/org/apache/hadoop/contrib/bkjournal/TestBookKeeperJournalManager.java

@@ -28,6 +28,7 @@ import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.ArrayList;
 import java.util.Random;
@@ -315,13 +316,13 @@ public class TestBookKeeperJournalManager {
     out.close();
     bkjm.finalizeLogSegment(1, numTransactions);
 
-     
+    List<EditLogInputStream> in = new ArrayList<EditLogInputStream>();
-    EditLogInputStream in = bkjm.getInputStream(1, true);
+    bkjm.selectInputStreams(in, 1, true);
     try {
       assertEquals(numTransactions, 
-                   FSEditLogTestUtil.countTransactionsInStream(in));
+                   FSEditLogTestUtil.countTransactionsInStream(in.get(0)));
     } finally {
-      in.close();
+      in.get(0).close();
     }
   }