hadoop-zookeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From maha...@apache.org
Subject svn commit: r903483 [6/6] - in /hadoop/zookeeper/trunk: ./ src/contrib/bookkeeper/src/java/org/apache/bookkeeper/bookie/ src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/ src/contrib/bookkeeper/src/java/org/apache/bookkeeper/proto/ src/cont...
Date Tue, 26 Jan 2010 23:16:49 GMT
Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java?rev=903483&r1=903482&r2=903483&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java
(original)
+++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java
Tue Jan 26 23:16:45 2010
@@ -1,4 +1,5 @@
 package org.apache.bookkeeper.test;
+
 /*
  * 
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -20,275 +21,54 @@
  * 
  */
 
-
-import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
-
-import java.lang.InterruptedException;
-import java.io.File;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import junit.framework.TestCase;
-
 import org.junit.*;
-import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
 import org.apache.log4j.Logger;
 
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.server.NIOServerCnxn;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.server.ServerStats;
-import org.apache.zookeeper.test.ClientBase;
-
 /**
- * This unit test tests closing ledgers sequentially. 
- * It creates 4 ledgers, then write 1000 entries to each 
- * ledger and close it.
+ * This unit test tests closing ledgers sequentially. It creates 4 ledgers, then
+ * write 1000 entries to each ledger and close it.
  * 
  */
 
-public class CloseTest 
-extends TestCase 
-implements Watcher {
+public class CloseTest extends BaseTestCase{
     static Logger LOG = Logger.getLogger(LedgerRecoveryTest.class);
-    
-    BookieServer bs1, bs2, bs3;
-    File tmpDir1, tmpDir2, tmpDir3, tmpDirZK;
-    private static final String HOSTPORT = "127.0.0.1:33299";
-    private NIOServerCnxn.Factory serverFactory;
-    
-    private static String BOOKIEADDR1 = "127.0.0.1:33300";
-    private static String BOOKIEADDR2 = "127.0.0.1:33301";
-    private static String BOOKIEADDR3 = "127.0.0.1:33302";
-    
-    private static void recursiveDelete(File dir) {
-        File children[] = dir.listFiles();
-        if (children != null) {
-            for(File child: children) {
-                recursiveDelete(child);
-            }
-        }
-        dir.delete();
-    }
-    
-    protected void setUp() throws Exception {
-        /*
-         * Creates 3 BookieServers
-         */
-        
-        
-        tmpDir1 = File.createTempFile("bookie1", "test");
-        tmpDir1.delete();
-        tmpDir1.mkdir();
-        
-        final int PORT1 = Integer.parseInt(BOOKIEADDR1.split(":")[1]);
-        bs1 = new BookieServer(PORT1, tmpDir1, new File[] { tmpDir1 });
-        bs1.start();
-        
-        tmpDir2 = File.createTempFile("bookie2", "test");
-        tmpDir2.delete();
-        tmpDir2.mkdir();
-        
-        final int PORT2 = Integer.parseInt(BOOKIEADDR2.split(":")[1]);
-        bs2 = new BookieServer(PORT2, tmpDir2, new File[] { tmpDir2 });
-        bs2.start();
-        
-        tmpDir3 = File.createTempFile("bookie3", "test");
-        tmpDir3.delete();
-        tmpDir3.mkdir();
-        
-        final int PORT3 = Integer.parseInt(BOOKIEADDR3.split(":")[1]);
-        bs3 = new BookieServer(PORT3, tmpDir3, new File[] { tmpDir3 });
-        bs3.start();
-        
-        /*
-         * Instantiates a ZooKeeper server. This is a blind copy
-         * of setUp from SessionTest.java.
-         */
-        LOG.info("STARTING " + getName());
-
-        //ServerStats.registerAsConcrete();
-
-        tmpDirZK = ClientBase.createTmpDir();
+    DigestType digestType;
 
-        ClientBase.setupTestEnv();
-        ZooKeeperServer zs = new ZooKeeperServer(tmpDirZK, tmpDirZK, 3000);
-        
-        final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        serverFactory = new NIOServerCnxn.Factory(PORT);
-        serverFactory.startup(zs);
-
-        assertTrue("waiting for server up",
-                   ClientBase.waitForServerUp(HOSTPORT,
-                                              CONNECTION_TIMEOUT));
-        
-        /*
-         * Creating necessary znodes
-         */
-        try{
-            ZooKeeper zk = new ZooKeeper(HOSTPORT, 3000, this);
-            zk.create("/ledgers", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-            zk.create("/ledgers/available", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-            zk.create("/ledgers/available/" + BOOKIEADDR1, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT );
-            zk.create("/ledgers/available/" + BOOKIEADDR2, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT );
-            zk.create("/ledgers/available/" + BOOKIEADDR3, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT );
-            zk.close();
-        } catch (KeeperException ke) {
-            LOG.error(ke);
-            fail("Couldn't execute ZooKeeper start procedure");
-        }
-        
-    }
-    
-    /**
-     * Watcher method. 
-     */
-    synchronized public void process(WatchedEvent event) {
-        LOG.info("Process: " + event.getType() + " " + event.getPath());
-    }
-    
-    protected void tearDown() throws Exception {
-        LOG.info("### Tear down ###");
-        bs1.shutdown();
-        recursiveDelete(tmpDir1);
-        
-        bs2.shutdown();
-        recursiveDelete(tmpDir2);
-        
-        bs3.shutdown();
-        recursiveDelete(tmpDir3);
-        
-        serverFactory.shutdown();
-        assertTrue("waiting for server down",
-                   ClientBase.waitForServerDown(HOSTPORT,
-                                                CONNECTION_TIMEOUT));
-
-        //ServerStats.unregister();
-        recursiveDelete(tmpDirZK);
-        LOG.info("FINISHED " + getName());
+    public CloseTest(DigestType digestType) {
+        super(3);
+        this.digestType = digestType;
     }
 
     @Test
-    public void testClose(){
-        /*
-         * 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");
-        }
-        
+    public void testClose() throws Exception {
+
         /*
          * Create 4 ledgers.
          */
-        LedgerHandle lh1 = null;
-        LedgerHandle lh2 = null;
-        LedgerHandle lh3 = null;
-        LedgerHandle lh4 = null;
-        
-        try{
-            lh1 = bk.createLedger("".getBytes());
-            lh2 = bk.createLedger("".getBytes());
-            lh3 = bk.createLedger("".getBytes());
-            lh4 = 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 1000 entries to lh1.
-         */
-        try{
-            String tmp = "BookKeeper is cool!";
-            for(int i = 0; i < 1000; i++){
-                lh1.addEntry(tmp.getBytes());
-            }
-        } catch(InterruptedException e){
-            LOG.error("Interrupted when adding entry", e);
-            fail("Couldn't finish adding entries");
-        } catch(BKException e){
-            LOG.error("BookKeeper exception", e);
-            fail("BookKeeper exception when adding entries");
-        }
-        
-        try{
-            lh1.close();
-        } catch(Exception e) {
-            LOG.error(e);
-            fail("Exception while closing ledger 1");
-        }
-        /*
-         * Write a 1000 entries to lh2.
-         */
-        try{
-            String tmp = "BookKeeper is cool!";
-            for(int i = 0; i < 1000; i++){
-                lh2.addEntry(tmp.getBytes());
-            }
-        } catch(InterruptedException e){
-            LOG.error("Interrupted when adding entry", e);
-            fail("Couldn't finish adding entries");
-        } catch(BKException e){
-            LOG.error("BookKeeper exception", e);
-            fail("CBookKeeper exception while adding entries");
-        }
-        
-        try{
-            lh2.close();
-        } catch(Exception e){
-            LOG.error(e);
-            fail("Exception while closing ledger 2");
+        int numLedgers = 4;
+        int numMsgs = 100;
+
+        LedgerHandle[] lh = new LedgerHandle[numLedgers];
+        for (int i = 0; i < numLedgers; i++) {
+            lh[i] = bkc.createLedger(digestType, "".getBytes());
         }
-        
+
+        String tmp = "BookKeeper is cool!";
+
         /*
-         * Write a 1000 entries to lh3 and lh4.
+         * Write 1000 entries to lh1.
          */
-        try{
-            String tmp = "BookKeeper is cool!";
-            for(int i = 0; i < 1000; i++){
-                lh3.addEntry(tmp.getBytes());
-                lh4.addEntry(tmp.getBytes());
+        for (int i = 0; i < numMsgs; i++) {
+            for (int j = 0; j < numLedgers; j++) {
+                lh[j].addEntry(tmp.getBytes());
             }
-        } catch(InterruptedException e){
-            LOG.error("Interrupted when adding entry", e);
-            fail("Couldn't finish adding entries");
-        } catch(BKException e){
-            LOG.error("BookKeeper exception", e);
-            fail("BookKeeper exception when adding entries");
         }
-        
-        try{
-            lh3.close();
-            lh4.close();
-        } catch(Exception e){
-            LOG.error(e);
-            fail("Exception while closing ledger 4");
+
+        for (int i = 0; i < numLedgers; i++) {
+
+            lh[i].close();
         }
-    }      
+    }
 }
-    
-    
\ No newline at end of file

Added: hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/ConcurrentLedgerTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/ConcurrentLedgerTest.java?rev=903483&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/ConcurrentLedgerTest.java
(added)
+++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/ConcurrentLedgerTest.java
Tue Jan 26 23:16:45 2010
@@ -0,0 +1,178 @@
+/*
+ *
+ * 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.test;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+/**
+ * Tests writing to concurrent ledgers
+ */
+public class ConcurrentLedgerTest extends TestCase {
+    Bookie bookie;
+    File txnDir, ledgerDir;
+    int recvTimeout = 10000;
+    Semaphore throttle;
+    
+    @Override
+    @Before
+    public void setUp() throws IOException {
+        String txnDirName = System.getProperty("txnDir");
+        if (txnDirName != null) {
+            txnDir = new File(txnDirName);
+        }
+        String ledgerDirName = System.getProperty("ledgerDir");
+        if (ledgerDirName != null) {
+            ledgerDir = new File(ledgerDirName);
+        }
+        File tmpFile = File.createTempFile("book", ".txn", txnDir);
+        tmpFile.delete();
+        txnDir = new File(tmpFile.getParent(), tmpFile.getName()+".dir");
+        txnDir.mkdirs();
+        tmpFile = File.createTempFile("book", ".ledger", ledgerDir);
+        ledgerDir = new File(tmpFile.getParent(), tmpFile.getName()+".dir");
+        ledgerDir.mkdirs();
+        
+        bookie = new Bookie(txnDir, new File[] {ledgerDir});
+    }
+    
+    static void recursiveDelete(File f) {
+        if (f.isFile()) {
+            f.delete();
+        } else {
+            for(File i: f.listFiles()) {
+                recursiveDelete(i);
+            }
+            f.delete();
+        }
+    }
+    
+    @Override
+    @After
+    public void tearDown() {
+        try {
+            bookie.shutdown();
+            recursiveDelete(txnDir);
+            recursiveDelete(ledgerDir);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    byte zeros[] = new byte[16];
+
+    int iterations = 51;
+    {
+        String iterationsString = System.getProperty("iterations");
+        if (iterationsString != null) {
+            iterations = Integer.parseInt(iterationsString);
+        }
+    }
+    int iterationStep = 25;
+    {
+        String iterationsString = System.getProperty("iterationStep");
+        if (iterationsString != null) {
+            iterationStep = Integer.parseInt(iterationsString);
+        }
+    }
+    @Test
+    public void testConcurrentWrite() throws IOException, InterruptedException, BookieException
{
+        int size = 1024;
+        int totalwrites = 128;
+        if (System.getProperty("totalwrites") != null) {
+            totalwrites = Integer.parseInt(System.getProperty("totalwrites"));
+        }
+        System.out.println("Running up to " + iterations + " iterations");
+        System.out.println("Total writes = " + totalwrites);
+        int ledgers;
+        for(ledgers = 1; ledgers <= iterations; ledgers += iterationStep) {
+            long duration = doWrites(ledgers, size, totalwrites);
+            System.out.println(totalwrites + " on " + ledgers + " took " + duration + " ms");
+        }
+        System.out.println("ledgers " + ledgers);
+        for(ledgers = 1; ledgers <= iterations; ledgers += iterationStep) {
+            long duration = doReads(ledgers, size, totalwrites);
+            System.out.println(ledgers + " read " + duration + " ms");
+        }
+    }
+
+    private long doReads(int ledgers, int size, int totalwrites)
+            throws IOException, InterruptedException, BookieException {
+        long start = System.currentTimeMillis();
+        for(int i = 1; i <= totalwrites/ledgers; i++) {
+            for(int j = 1; j <= ledgers; j++) {
+                ByteBuffer entry = bookie.readEntry(j, i);
+                // skip the ledger id and the entry id
+                entry.getLong();
+                entry.getLong();
+                assertEquals(j + "@" + i, j+2, entry.getLong());
+                assertEquals(j + "@" + i, i+3, entry.getLong());
+            }
+        }
+        long finish = System.currentTimeMillis();
+        return finish - start;
+    }
+    private long doWrites(int ledgers, int size, int totalwrites)
+            throws IOException, InterruptedException, BookieException {
+        throttle = new Semaphore(10000);
+        WriteCallback cb = new WriteCallback() {
+            @Override
+            public void writeComplete(int rc, long ledgerId, long entryId,
+                    InetSocketAddress addr, Object ctx) {
+                AtomicInteger counter = (AtomicInteger)ctx;
+                counter.getAndIncrement();
+                throttle.release();
+            }
+        };
+        AtomicInteger counter = new AtomicInteger();
+        long start = System.currentTimeMillis();
+        for(int i = 1; i <= totalwrites/ledgers; i++) {
+            for(int j = 1; j <= ledgers; j++) {
+                ByteBuffer bytes = ByteBuffer.allocate(size);
+                bytes.putLong(j);
+                bytes.putLong(i);
+                bytes.putLong(j+2);
+                bytes.putLong(i+3);
+                bytes.put(("This is ledger " + j + " entry " + i).getBytes());
+                bytes.position(0);
+                bytes.limit(bytes.capacity());
+                throttle.acquire();
+                bookie.addEntry(bytes, cb, counter, zeros);
+            }
+        }
+        long finish = System.currentTimeMillis();
+        return finish - start;
+    }
+}

Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LedgerRecoveryTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LedgerRecoveryTest.java?rev=903483&r1=903482&r2=903483&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LedgerRecoveryTest.java
(original)
+++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LedgerRecoveryTest.java
Tue Jan 26 23:16:45 2010
@@ -1,4 +1,5 @@
 package org.apache.bookkeeper.test;
+
 /*
  * 
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -20,318 +21,65 @@
  * 
  */
 
-
-import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
-
-import java.lang.InterruptedException;
-import java.io.File;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import junit.framework.TestCase;
-
 import org.junit.*;
-import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
 import org.apache.log4j.Logger;
 
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.server.NIOServerCnxn;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.server.ServerStats;
-import org.apache.zookeeper.test.ClientBase;
-
 /**
- * This unit test tests ledger recovery. 
+ * This unit test tests ledger recovery.
+ * 
  * 
- *
  */
 
-public class LedgerRecoveryTest 
-extends TestCase 
-implements Watcher {
+public class LedgerRecoveryTest extends BaseTestCase {
     static Logger LOG = Logger.getLogger(LedgerRecoveryTest.class);
-    
-    BookieServer bs1, bs2, bs3;
-    File tmpDir1, tmpDir2, tmpDir3, tmpDirZK;
-    private static final String HOSTPORT = "127.0.0.1:33299";
-    private NIOServerCnxn.Factory serverFactory;
-    
-    private static String BOOKIEADDR1 = "127.0.0.1:33300";
-    private static String BOOKIEADDR2 = "127.0.0.1:33301";
-    private static String BOOKIEADDR3 = "127.0.0.1:33302";
-    
-    private static void recursiveDelete(File dir) {
-        File children[] = dir.listFiles();
-        if (children != null) {
-            for(File child: children) {
-                recursiveDelete(child);
-            }
-        }
-        dir.delete();
-    }
-    
-    protected void setUp() throws Exception {
-        /*
-         * Creates 3 BookieServers
-         */
-        
-        
-        tmpDir1 = File.createTempFile("bookie1", "test");
-        tmpDir1.delete();
-        tmpDir1.mkdir();
-        
-        final int PORT1 = Integer.parseInt(BOOKIEADDR1.split(":")[1]);
-        bs1 = new BookieServer(PORT1, tmpDir1, new File[] { tmpDir1 });
-        bs1.start();
-        
-        tmpDir2 = File.createTempFile("bookie2", "test");
-        tmpDir2.delete();
-        tmpDir2.mkdir();
-        
-        final int PORT2 = Integer.parseInt(BOOKIEADDR2.split(":")[1]);
-        bs2 = new BookieServer(PORT2, tmpDir2, new File[] { tmpDir2 });
-        bs2.start();
-        
-        tmpDir3 = File.createTempFile("bookie3", "test");
-        tmpDir3.delete();
-        tmpDir3.mkdir();
-        
-        final int PORT3 = Integer.parseInt(BOOKIEADDR3.split(":")[1]);
-        bs3 = new BookieServer(PORT3, tmpDir3, new File[] { tmpDir3 });
-        bs3.start();
-        
-        /*
-         * Instantiates a ZooKeeper server. This is a blind copy
-         * of setUp from SessionTest.java.
-         */
-        LOG.info("STARTING " + getName());
-
-        //ServerStats.registerAsConcrete();
 
-        tmpDirZK = ClientBase.createTmpDir();
+    DigestType digestType;
 
-        ClientBase.setupTestEnv();
-        ZooKeeperServer zs = new ZooKeeperServer(tmpDirZK, tmpDirZK, 3000);
-        
-        final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        serverFactory = new NIOServerCnxn.Factory(PORT);
-        serverFactory.startup(zs);
-
-        assertTrue("waiting for server up",
-                   ClientBase.waitForServerUp(HOSTPORT,
-                                              CONNECTION_TIMEOUT));
-        
-        /*
-         * Creating necessary znodes
-         */
-        try{
-            ZooKeeper zk = new ZooKeeper(HOSTPORT, 3000, this);
-            zk.create("/ledgers", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-            zk.create("/ledgers/available", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-            zk.create("/ledgers/available/" + BOOKIEADDR1, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT );
-            zk.create("/ledgers/available/" + BOOKIEADDR2, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT );
-            zk.create("/ledgers/available/" + BOOKIEADDR3, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT );
-            zk.close();
-        } catch (KeeperException ke) {
-            LOG.error(ke);
-            fail("Couldn't execute ZooKeeper start procedure");
-        }
-        
-    }
-    
-    /**
-     * Watcher method. 
-     */
-    synchronized public void process(WatchedEvent event) {
-        LOG.info("Process: " + event.getType() + " " + event.getPath());
+    public LedgerRecoveryTest(DigestType digestType) {
+        super(3);
+        this.digestType = digestType;
     }
-    
-    protected void tearDown() throws Exception {
-        LOG.info("### Tear down ###");
-        bs1.shutdown();
-        recursiveDelete(tmpDir1);
-        
-        bs2.shutdown();
-        recursiveDelete(tmpDir2);
-        
-        bs3.shutdown();
-        recursiveDelete(tmpDir3);
-        
-        serverFactory.shutdown();
-        assertTrue("waiting for server down",
-                   ClientBase.waitForServerDown(HOSTPORT,
-                                                CONNECTION_TIMEOUT));
-
-        //ServerStats.unregister();
-        recursiveDelete(tmpDirZK);
-        LOG.info("FINISHED " + getName());
-    }
-    
-    @Test
-    public void testLedgerRecovery(){
-        /*
-         * 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");
-        }
-        
+
+    private void testInternal(int numEntries) throws Exception {
         /*
          * 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");
+        beforelh = bkc.createLedger(digestType, "".getBytes());
+
+        String tmp = "BookKeeper is cool!";
+        for (int i = 0; i < numEntries; i++) {
+            beforelh.addEntry(tmp.getBytes());
         }
-        
+
         /*
-         * Write a 1000 entries.
+         * Try to open ledger.
          */
-        try{
-            String tmp = "BookKeeper is cool!";
-            for(int i = 0; i < 1000; i++){
-                beforelh.addEntry(tmp.getBytes());
-            }
-            
-            //bk.resetLedger(beforelh);
-        } catch(InterruptedException e){
-            LOG.error("Interrupted when adding entry", e);
-            fail("Couldn't finish adding entries");
-        } catch(BKException e){
-            LOG.error("BookKeeper exception", e);
-            fail("BookKeeper exception while adding entries");
-        }
-        
+        LedgerHandle afterlh = bkc.openLedger(beforelh.getId(), digestType, "".getBytes());
+
         /*
-         * Try to open ledger.
+         * Check if has recovered properly.
          */
-        try{
-            LedgerHandle afterlh = bk.openLedger(beforelh.getId(), "".getBytes());
-            
-            /*
-             * Check if has recovered properly.
-             */
-            assertTrue("Has not recovered correctly: " + afterlh.getLast(), afterlh.getLast()
== 999);
-        } 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");
-        }
-        
+        assertTrue("Has not recovered correctly: " + afterlh.getLastAddConfirmed(),
+                afterlh.getLastAddConfirmed() == numEntries - 1);        
     }
     
     @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++){
-                beforelh.addEntry(tmp.getBytes());
-            }
-        } catch(InterruptedException e){
-            LOG.error("Interrupted when adding entry", e);
-            fail("Couldn't finish adding entries");
-        } catch(BKException e){
-            LOG.error("BookKeeper exception", e);
-            fail("BookKeeper exception while adding entries");
-        }
-        
-        
-        /*
-         * 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()
== 0);
-        } 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");
-        }
-        
+    public void testLedgerRecovery() throws Exception {
+        testInternal(100);
+     
     }
-    
+
+    @Test
+    public void testEmptyLedgerRecoveryOne() throws Exception{
+        testInternal(1);
+    }
+
+    @Test
+    public void testEmptyLedgerRecovery() throws Exception{
+        testInternal(0);
+    }
+
 }

Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LoopbackClient.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LoopbackClient.java?rev=903483&r1=903482&r2=903483&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LoopbackClient.java
(original)
+++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/LoopbackClient.java
Tue Jan 26 23:16:45 2010
@@ -1,4 +1,5 @@
 package org.apache.bookkeeper.test;
+
 /*
  * 
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -20,103 +21,97 @@
  * 
  */
 
-
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
 import java.io.IOException;
 import java.lang.InterruptedException;
 import java.util.Arrays;
+import java.util.concurrent.Executors;
 
 import org.apache.bookkeeper.proto.BookieClient;
-import org.apache.bookkeeper.proto.WriteCallback;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
+import org.apache.bookkeeper.util.OrderedSafeExecutor;
 import org.apache.log4j.Logger;
-
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
 
 /**
  * This class tests BookieClient. It just sends the a new entry to itself.
  * 
  * 
- *
+ * 
  */
 
-
 class LoopbackClient implements WriteCallback {
     Logger LOG = Logger.getLogger(LoopbackClient.class);
     BookieClient client;
     static int recvTimeout = 2000;
     long begin = 0;
     int limit;
-    
-    
+    OrderedSafeExecutor executor;
+
     static class Counter {
         int c;
         int limit;
-        
-        Counter(int limit){
+
+        Counter(int limit) {
             this.c = 0;
             this.limit = limit;
         }
-        
-        synchronized void increment(){
-            if(++c == limit) 
+
+        synchronized void increment() {
+            if (++c == limit)
                 this.notify();
         }
     }
-    
-    LoopbackClient(int port, long begin, int limit)
-    throws IOException {
-        this.client = 
-            new BookieClient(new InetSocketAddress("127.0.0.1", port), recvTimeout);
+
+    LoopbackClient(ClientSocketChannelFactory channelFactory, OrderedSafeExecutor executor,
long begin, int limit) throws IOException {
+        this.client = new BookieClient(channelFactory, executor);
         this.begin = begin;
     }
-    
-    
-    void write(long ledgerId, long entry, byte[] data, WriteCallback cb, Object ctx)
-    throws IOException, InterruptedException {
+
+    void write(long ledgerId, long entry, byte[] data, InetSocketAddress addr, WriteCallback
cb, Object ctx)
+            throws IOException, InterruptedException {
         LOG.info("Ledger id: " + ledgerId + ", Entry: " + entry);
         byte[] passwd = new byte[20];
         Arrays.fill(passwd, (byte) 'a');
-        
-        client.addEntry(ledgerId, 
-            passwd,
-            entry, 
-            ByteBuffer.wrap(data), 
-            cb,
-            ctx);
+
+        client.addEntry(addr, ledgerId, passwd, entry, ChannelBuffers.wrappedBuffer(data),
cb, ctx);
     }
-    
-    public void writeComplete(int rc, long ledgerId, long entryId, Object ctx){
+
+    public void writeComplete(int rc, long ledgerId, long entryId, InetSocketAddress addr,
Object ctx) {
         Counter counter = (Counter) ctx;
         counter.increment();
     }
-    
-    
-    public static void main(String args[]){
+
+    public static void main(String args[]) {
         byte[] data = new byte[Integer.parseInt(args[0])];
         Integer limit = Integer.parseInt(args[1]);
         Counter c = new Counter(limit);
         long ledgerId = Long.valueOf("0").longValue();
         long begin = System.currentTimeMillis();
-        
+
         LoopbackClient lb;
-        try{
-            lb = new LoopbackClient(Integer.valueOf(args[2]).intValue(), 
-                    begin, 
-                    limit.intValue());
-        
-            for(int i = 0; i < limit ; i++){
-                lb.write(ledgerId, i, data, lb, c);   
+        ClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),
Executors
+                .newCachedThreadPool());
+        OrderedSafeExecutor executor = new OrderedSafeExecutor(2);
+        try {
+            InetSocketAddress addr = new InetSocketAddress("127.0.0.1", Integer.valueOf(args[2]).intValue());
+            lb = new LoopbackClient(channelFactory, executor, begin, limit.intValue());
+
+            for (int i = 0; i < limit; i++) {
+                lb.write(ledgerId, i, data, addr, lb, c);
             }
-            
-            synchronized(c){
+
+            synchronized (c) {
                 c.wait();
                 System.out.println("Time to write all entries: " + (System.currentTimeMillis()
- begin));
             }
-        } catch (IOException e){
+        } catch (IOException e) {
             e.printStackTrace();
-        } catch (InterruptedException e){
+        } catch (InterruptedException e) {
             e.printStackTrace();
         }
-    } 
-    
+    }
+
 }

Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/NIOServerFactoryTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/NIOServerFactoryTest.java?rev=903483&r1=903482&r2=903483&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/NIOServerFactoryTest.java
(original)
+++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/NIOServerFactoryTest.java
Tue Jan 26 23:16:45 2010
@@ -1,4 +1,5 @@
 package org.apache.bookkeeper.test;
+
 /*
  * 
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -20,7 +21,6 @@
  * 
  */
 
-
 import java.net.Socket;
 import java.nio.ByteBuffer;
 
@@ -29,7 +29,6 @@
 import org.apache.bookkeeper.proto.NIOServerFactory.PacketProcessor;
 import org.junit.Test;
 
-
 import junit.framework.TestCase;
 
 public class NIOServerFactoryTest extends TestCase {
@@ -41,8 +40,9 @@
             }
             src.sendResponse(new ByteBuffer[] { ByteBuffer.allocate(4) });
         }
-        
+
     };
+
     @Test
     public void testProblemProcessor() throws Exception {
         NIOServerFactory factory = new NIOServerFactory(22334, problemProcessor);



Mime
View raw message