ZOOKEEPER-465. Ledger size in bytes. (flavio via mahadev)

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

CHANGES.txt

@@ -234,6 +234,8 @@ IMPROVEMENTS:
 
   ZOOKEEPER-977. passing null for path_buffer in zoo_create (breed via mahadev)
 
+  ZOOKEEPER-465. Ledger size in bytes. (flavio via mahadev)
+
 NEW FEATURES:
   ZOOKEEPER-729. Java client API to recursively delete a subtree.
   (Kay Kay via henry)

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

@@ -37,7 +37,9 @@ import org.jboss.netty.buffer.ChannelBuffers;
 
 public abstract class DigestManager {
     static final Logger logger = Logger.getLogger(DigestManager.class);
-
+    
+    static final int METADATA_LENGTH = 32;
+    
     long ledgerId;
     
     abstract int getMacCodeLength();
@@ -67,21 +69,32 @@ public abstract class DigestManager {
         }
     }
 
-    public ChannelBuffer computeDigestAndPackageForSending(long entryId, long lastAddConfirmed, byte[] data) {
+    /**
+     * Computes the digest for an entry and put bytes together for sending.
+     *  
+     * @param entryId
+     * @param lastAddConfirmed
+     * @param length
+     * @param data
+     * @return
+     */
+    
+    public ChannelBuffer computeDigestAndPackageForSending(long entryId, long lastAddConfirmed, long length, byte[] data) {
 
-        byte[] bufferArray = new byte[24+macCodeLength];
+        byte[] bufferArray = new byte[METADATA_LENGTH + macCodeLength];
         ByteBuffer buffer = ByteBuffer.wrap(bufferArray);
         buffer.putLong(ledgerId);
         buffer.putLong(entryId);
         buffer.putLong(lastAddConfirmed);
+        buffer.putLong(length);
         buffer.flip();
 
-        update(buffer.array(), 0, 24);
+        update(buffer.array(), 0, METADATA_LENGTH);
         update(data);
         byte[] digest = getValueAndReset();
 
         buffer.limit(buffer.capacity());
-        buffer.position(24);
+        buffer.position(METADATA_LENGTH);
         buffer.put(digest);
         buffer.flip();
 
@@ -102,14 +115,14 @@ public abstract class DigestManager {
         ByteBuffer dataReceivedBuffer = dataReceived.toByteBuffer();
         byte[] digest;
 
-        update(dataReceivedBuffer.array(), dataReceivedBuffer.position(), 24);
+        update(dataReceivedBuffer.array(), dataReceivedBuffer.position(), METADATA_LENGTH);
 
-        int offset = 24 + macCodeLength;
+        int offset = METADATA_LENGTH + macCodeLength;
         update(dataReceivedBuffer.array(), dataReceivedBuffer.position() + offset, dataReceived.readableBytes() - offset);
         digest = getValueAndReset();
 
         for (int i = 0; i < digest.length; i++) {
-            if (digest[i] != dataReceived.getByte(24 + i)) {
+            if (digest[i] != dataReceived.getByte(METADATA_LENGTH + i)) {
                 logger.error("Mac mismatch for ledger-id: " + ledgerId + ", entry-id: " + entryId);
                 throw new BKDigestMatchException();
             }
@@ -131,11 +144,19 @@ public abstract class DigestManager {
         }
 
     }
-
+    
+    /**
+     * Verify that the digest matches and returns the data in the entry.
+     * 
+     * @param entryId
+     * @param dataReceived
+     * @return
+     * @throws BKDigestMatchException
+     */
     ChannelBufferInputStream verifyDigestAndReturnData(long entryId, ChannelBuffer dataReceived)
             throws BKDigestMatchException {
         verifyDigest(entryId, dataReceived);
-        dataReceived.readerIndex(24 + macCodeLength);
+        dataReceived.readerIndex(METADATA_LENGTH + macCodeLength);
         return new ChannelBufferInputStream(dataReceived);
     }
 
@@ -156,6 +177,7 @@ public abstract class DigestManager {
 
         long entryId = dataReceived.readLong();
         long lastAddConfirmed = dataReceived.readLong();
+        long length = dataReceived.readLong();
         return new RecoveryData(lastAddConfirmed, entryId);
 
     }

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

@@ -38,6 +38,7 @@ public class LedgerEntry {
 
   long ledgerId;
   long entryId;
+  long length;
   ChannelBufferInputStream entryDataStream;
 
   int nextReplicaIndexToReadFrom = 0;
@@ -54,6 +55,10 @@ public class LedgerEntry {
   public long getEntryId() {
     return entryId;
   }
+  
+  public long getLength() {
+      return length;
+  }
 
   public byte[] getEntry() {
     try {

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

@@ -60,6 +60,7 @@ public class LedgerHandle implements ReadCallback, AddCallback, CloseCallback {
   final long ledgerId;
   long lastAddPushed;
   long lastAddConfirmed;
+  long length;
   final DigestManager macManager;
   final DistributionSchedule distributionSchedule;
 
@@ -75,8 +76,10 @@ public class LedgerHandle implements ReadCallback, AddCallback, CloseCallback {
     this.metadata = metadata;
     if (metadata.isClosed()) {
       lastAddConfirmed = lastAddPushed = metadata.close;
+      length = metadata.length;
     } else {
       lastAddConfirmed = lastAddPushed = -1;
+      length = 0;
     }
     
     this.ledgerId = ledgerId;
@@ -157,6 +160,26 @@ public class LedgerHandle implements ReadCallback, AddCallback, CloseCallback {
       return this.opCounterSem;
   }
   
+  /**
+   *  Add to the length of the ledger in bytes.
+   *  
+   * @param delta
+   * @return
+   */
+  long addToLength(long delta){
+      this.length += delta;
+      return this.length;
+  }
+  
+  /**
+   * Returns the length of the ledger in bytes. 
+   * 
+   * @return
+   */
+  public long getLength(){
+      return this.length;
+  }
+  
   /**
    * Get the Distribution Schedule
    * 
@@ -212,6 +235,7 @@ public class LedgerHandle implements ReadCallback, AddCallback, CloseCallback {
 
       @Override
       public void safeRun() {
+        metadata.length = length;
         // Close operation is idempotent, so no need to check if we are
         // already closed
         metadata.close(lastAddConfirmed);
@@ -220,7 +244,7 @@ public class LedgerHandle implements ReadCallback, AddCallback, CloseCallback {
 
         if (LOG.isDebugEnabled()) {
           LOG.debug("Closing ledger: " + ledgerId + " at entryId: "
-              + metadata.close);
+              + metadata.close + " with this many bytes: " + metadata.length);
         }
 
         writeLedgerConfig(new StatCallback() {
@@ -343,10 +367,11 @@ public class LedgerHandle implements ReadCallback, AddCallback, CloseCallback {
                   }
 
                   long entryId = ++lastAddPushed;
+                  long currentLength = addToLength(data.length);
                   PendingAddOp op = new PendingAddOp(LedgerHandle.this, cb, ctx, entryId);
                   pendingAddOps.add(op);
                   ChannelBuffer toSend = macManager.computeDigestAndPackageForSending(
-                          entryId, lastAddConfirmed, data);
+                          entryId, lastAddConfirmed, currentLength, data);
                   op.initiate(toSend);
               }
           });

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

@@ -45,6 +45,7 @@ public class LedgerMetadata {
     public static final int NOTCLOSED = -101;
     int ensembleSize;
     int quorumSize;
+    long length;
     long close;
     private SortedMap<Long, ArrayList<InetSocketAddress>> ensembles = new TreeMap<Long, ArrayList<InetSocketAddress>>();
     ArrayList<InetSocketAddress> currentEnsemble;
@@ -52,6 +53,12 @@ public class LedgerMetadata {
     public LedgerMetadata(int ensembleSize, int quorumSize) {
         this.ensembleSize = ensembleSize;
         this.quorumSize = quorumSize;
+        
+        /*
+         * It is set in PendingReadOp.readEntryComplete, and 
+         * we read it in LedgerRecoveryOp.readComplete.
+         */
+        this.length = 0;
         this.close = NOTCLOSED;
     };
 
@@ -77,7 +84,7 @@ public class LedgerMetadata {
     void close(long entryId) {
         close = entryId;
     }
-
+    
     void addEnsemble(long startEntryId, ArrayList<InetSocketAddress> ensemble) {
         assert ensembles.isEmpty() || startEntryId >= ensembles.lastKey();
 
@@ -117,7 +124,7 @@ public class LedgerMetadata {
      */
     public byte[] serialize() {
         StringBuilder s = new StringBuilder();
-        s.append(quorumSize).append(lSplitter).append(ensembleSize);
+        s.append(quorumSize).append(lSplitter).append(ensembleSize).append(lSplitter).append(length);
 
         for (Map.Entry<Long, ArrayList<InetSocketAddress>> entry : ensembles.entrySet()) {
             s.append(lSplitter).append(entry.getKey());
@@ -126,7 +133,7 @@ public class LedgerMetadata {
                 StringUtils.addrToString(s, addr);
             }
         }
-
+        
         if (close != NOTCLOSED) {
             s.append(lSplitter).append(close).append(tSplitter).append(closed);
         }
@@ -166,8 +173,9 @@ public class LedgerMetadata {
         try {
             lc.quorumSize = new Integer(lines[0]);
             lc.ensembleSize = new Integer(lines[1]);
-
-            for (int i = 2; i < lines.length; i++) {
+            lc.length = new Long(lines[2]); 
+            
+            for (int i = 3; i < lines.length; i++) {
                 String parts[] = lines[i].split(tSplitter);
 
                 if (parts[1].equals(closed)) {

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

@@ -44,6 +44,7 @@ class LedgerRecoveryOp implements ReadEntryCallback, ReadCallback, AddCallback {
     boolean proceedingWithRecovery = false;
     long maxAddPushed = -1;
     long maxAddConfirmed = -1;
+    long maxLength = 0;
 
     GenericCallback<Void> cb;
 
@@ -81,8 +82,7 @@ class LedgerRecoveryOp implements ReadEntryCallback, ReadCallback, AddCallback {
                 heardValidResponse = true;
             } catch (BKDigestMatchException e) {
                 // Too bad, this bookie didnt give us a valid answer, we
-                // still
-                // might be able to recover though so continue
+                // still might be able to recover though so continue
                 LOG.error("Mac mismatch while reading last entry from bookie: "
                         + lh.metadata.currentEnsemble.get(bookieIndex));
             }
@@ -99,6 +99,7 @@ class LedgerRecoveryOp implements ReadEntryCallback, ReadCallback, AddCallback {
         if (heardValidResponse && lh.distributionSchedule.canProceedWithRecovery(bookieIndex)) {
             proceedingWithRecovery = true;
             lh.lastAddPushed = lh.lastAddConfirmed = maxAddConfirmed;
+            lh.length = maxLength;
             doRecoveryRead();
             return;
         }
@@ -126,7 +127,17 @@ class LedgerRecoveryOp implements ReadEntryCallback, ReadCallback, AddCallback {
         // get back to prev value
         lh.lastAddConfirmed--;
         if (rc == BKException.Code.OK) {
-            lh.asyncAddEntry(seq.nextElement().getEntry(), this, null);
+            LedgerEntry entry = seq.nextElement(); 
+            byte[] data = entry.getEntry();
+            
+            /*
+             * We will add this entry again to make sure it is written to enough
+             * replicas. We subtract the length of the data itself, since it will
+             * be added again when processing the call to add it.
+             */
+            lh.length = entry.getLength() - (long) data.length;
+            lh.asyncAddEntry(data, this, null);
+            
             return;
         }
 
@@ -138,6 +149,7 @@ class LedgerRecoveryOp implements ReadEntryCallback, ReadCallback, AddCallback {
             // it
 
             cb.operationComplete(BKException.Code.OK, null);
+            LOG.debug("After closing length is: " + lh.getLength());
             return;
         }
 

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

@@ -22,6 +22,7 @@ package org.apache.bookkeeper.client;
  */
 
 import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Enumeration;
@@ -34,6 +35,8 @@ import org.apache.log4j.Logger;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBufferInputStream;
 
+import java.io.IOException;
+
 /**
  * Sequence of entries of a ledger that represents a pending read operation.
  * When all the data read has come back, the application callback is called.
@@ -126,8 +129,14 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback {
             logErrorAndReattemptRead(entry, "Mac mismatch", BKException.Code.DigestMatchException);
             return;
         }
-
+        
         entry.entryDataStream = is;
+        
+        /*
+         * The length is a long and it is the last field of the metadata of an entry.
+         * Consequently, we have to subtract 8 from METADATA_LENGTH to get the length.
+         */
+        entry.length = buffer.getLong(DigestManager.METADATA_LENGTH - 8);
 
         numPendingReads--;
         if (numPendingReads == 0) {

src/contrib/bookkeeper/src/java/org/apache/bookkeeper/tools/BookKeeperTools.java

@@ -686,8 +686,9 @@ public class BookKeeperTools {
                  * Now that we've read the ledger entry, write it to the new
                  * bookie we've selected.
                  */
+                LedgerEntry entry = seq.nextElement();
                 ChannelBuffer toSend = lh.getDigestManager().computeDigestAndPackageForSending(entryId,
-                        lh.getLastAddConfirmed(), seq.nextElement().getEntry());
+                        lh.getLastAddConfirmed(), entry.getLength(), entry.getEntry());
                 bkc.getBookieClient().addEntry(newBookie, lh.getId(), lh.getLedgerKey(), entryId, toSend,
                         new WriteCallback() {
                             @Override

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

@@ -139,7 +139,7 @@ public class AsyncLedgerOpsTest extends BaseTestCase implements AddCallback, Rea
                 }
             }
 
-            LOG.debug("*** WRITE COMPLETE ***");
+            LOG.info("*** WRITE COMPLETE ***");
             // close ledger
             synchronized (ctx) {
                 lh.asyncClose(this, ctx);

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

@@ -609,6 +609,60 @@ public class BookieReadWriteTest extends BaseTestCase implements AddCallback, Re
         }
     }
 
+    @Test
+    public void testReadWriteAsyncLength() throws IOException {
+        try {
+            // Create a BookKeeper client and a ledger
+            bkc = new BookKeeper("127.0.0.1");
+            lh = bkc.createLedger(digestType, ledgerPassword);
+            // bkc.initMessageDigest("SHA1");
+            ledgerId = lh.getId();
+            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);
+
+                entries.add(entry.array());
+                entriesSize.add(entry.array().length);
+                lh.asyncAddEntry(entry.array(), this, sync);
+            }
+
+            // wait for all entries to be acknowledged
+            synchronized (sync) {
+                while (sync.counter < numEntriesToWrite) {
+                    LOG.debug("Entries counter = " + sync.counter);
+                    sync.wait();
+                }
+            }
+            long length = numEntriesToWrite * 4;
+            assertTrue("Ledger length before closing: " + lh.getLength(), lh.getLength() == length);
+            
+            LOG.debug("*** WRITE COMPLETE ***");
+            // close ledger
+            lh.close();
+
+            // *** WRITING PART COMPLETE // READ PART BEGINS ***
+
+            // open ledger
+            lh = bkc.openLedger(ledgerId, digestType, ledgerPassword);
+            assertTrue("Ledger length after opening: " + lh.getLength(), lh.getLength() == length);
+
+
+            lh.close();
+        } catch (KeeperException e) {
+            LOG.error("Test failed", e);
+            fail("Test failed due to ZooKeeper exception");
+        } catch (BKException e) {
+            LOG.error("Test failed", e);
+            fail("Test failed due to BookKeeper exception");
+        } catch (InterruptedException e) {
+            LOG.error("Test failed", e);
+            fail("Test failed due to interruption");
+        }
+    }
+    
+    
     public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) {
         if(rc != BKException.Code.OK) fail("Return code is not OK: " + rc);
         

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

@@ -29,7 +29,6 @@ import org.apache.log4j.Logger;
 /**
  * This unit test tests ledger recovery.
  * 
- * 
  */
 
 public class LedgerRecoveryTest extends BaseTestCase {
@@ -54,6 +53,8 @@ public class LedgerRecoveryTest extends BaseTestCase {
             beforelh.addEntry(tmp.getBytes());
         }
 
+        long length = (long) (numEntries * tmp.length());
+        
         /*
          * Try to open ledger.
          */
@@ -63,7 +64,9 @@ public class LedgerRecoveryTest extends BaseTestCase {
          * Check if has recovered properly.
          */
         assertTrue("Has not recovered correctly: " + afterlh.getLastAddConfirmed(),
-                afterlh.getLastAddConfirmed() == numEntries - 1);        
+                afterlh.getLastAddConfirmed() == numEntries - 1);       
+        assertTrue("Has not set the length correctly: " + afterlh.getLength() + ", " + length, 
+                afterlh.getLength() == length);
     }
     
     @Test