ZOOKEEPER-363. NPE when recovering ledger with no hint. (flavio via mahadev)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/zookeeper/trunk@762602 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -39,6 +39,8 @@ BUGFIXES:
   ZOOKEEPER-362. Issues with FLENewEpochTest. (fix bug in Fast leader election)
 (flavio via mahadev)
 
+  ZOOKEEPER-363. NPE when recovering ledger with no hint. (flavio via mahadev)
+
 IMPROVEMENTS:
   ZOOKEEPER-308. improve the atomic broadcast performance 3x.
   (breed via mahadev)

src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/LedgerRecoveryMonitor.java

@@ -145,8 +145,8 @@ class LedgerRecoveryMonitor implements ReadEntryCallback{
         long readCounter = 0;
         while(notLegitimate){
             readCounter = getNextHint();
-            if(readCounter != -1){
-                lh.setLast(readCounter - 1);
+            if(readCounter > -1){
+                lh.setLast(readCounter);
                 boolean hasMore = true;
                 while(hasMore){
                     hasMore = false;

src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LedgerRecoveryTest.java

@@ -260,4 +260,89 @@ implements Watcher {
         
     }
     
+    @Test
+    public void testEmptyLedgerRecovery(){
+        /*
+         * Instantiate BookKeeper object.
+         */
+        BookKeeper bk = null;
+        try{
+            bk = new BookKeeper(HOSTPORT);
+        } catch (KeeperException ke){
+            LOG.error("Error instantiating BookKeeper", ke);
+            fail("ZooKeeper error");
+        } catch (IOException ioe){
+            LOG.error(ioe);
+            fail("Failure due to IOException");
+        }
+        
+        /*
+         * Create ledger.
+         */
+        LedgerHandle beforelh = null;
+        try{
+            beforelh = bk.createLedger("".getBytes());
+        } catch (KeeperException ke){
+            LOG.error("Error creating a ledger", ke);
+            fail("ZooKeeper error");            
+        } catch (BKException bke){
+            LOG.error("BookKeeper error");
+            fail("BookKeeper error");
+        } catch (InterruptedException ie) {
+            LOG.error(ie);
+            fail("Failure due to interrupted exception");
+        } catch (IOException ioe) {
+            LOG.error(ioe);
+            fail("Failure due to IO exception");
+        }
+        
+        /*
+         * Write a 1 entry.
+         */
+        try{
+            String tmp = "BookKeeper is cool!";
+            for(int i = 0; i < 1; i++){
+                bk.addEntry(beforelh, tmp.getBytes());
+            }
+        } catch(InterruptedException e){
+            LOG.error("Interrupted when adding entry", e);
+            fail("Couldn't finish adding entries");
+        }
+        
+        ///*
+        // * Sleep.
+        // */
+        //try{
+        //    Thread.sleep(2000);
+        //} catch(InterruptedException e){
+        //    LOG.error("Interrupted while sleeping", e);
+        //    fail("Couldn't finish sleeping");
+        //}
+        
+        /*
+         * Try to open ledger.
+         */
+        try{
+            LedgerHandle afterlh = bk.openLedger(beforelh.getId(), "".getBytes());
+            
+            /*
+             * Check if has recovered properly.
+             */
+            assertTrue("Has not recovered correctly: " + afterlh.getLast(), afterlh.getLast() == 1);
+        } catch (KeeperException e) {
+            LOG.error("Error when opening ledger", e);
+            fail("Couldn't open ledger");
+        } catch (InterruptedException ie) {
+            LOG.error("Interrupted exception", ie);
+            fail("Failure due to interrupted exception");
+        } catch (IOException ioe) {
+            LOG.error("IO Exception", ioe);
+            fail("Failure due to IO exception");
+        } catch (BKException bke){
+            LOG.error("BookKeeper error", bke);
+            fail("BookKeeper error");
+        }
+        
+    }
+    
 }