bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [04/49] bookkeeper git commit: BOOKKEEPER-769: Remove the Hedwig Code
Date Wed, 16 Mar 2016 03:44:14 GMT
http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHubSSL.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHubSSL.java b/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHubSSL.java
deleted file mode 100644
index 98d36b6..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigHubSSL.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- * 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.hedwig.server.integration;
-
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-import java.util.Collection;
-import java.util.Arrays;
-
-@RunWith(Parameterized.class)
-public class TestHedwigHubSSL extends TestHedwigHub {
-    @Parameters
-    public static Collection<Object[]> configs() {
-        return Arrays.asList(new Object[][] { { true }, { false } });
-    }
-
-    public TestHedwigHubSSL(boolean isSubscriptionChannelSharingEnabled) {
-        super(Mode.SSL, isSubscriptionChannelSharingEnabled);
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java b/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java
deleted file mode 100644
index bce41e5..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/integration/TestHedwigRegion.java
+++ /dev/null
@@ -1,304 +0,0 @@
-/**
- * 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.hedwig.server.integration;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Random;
-import java.util.concurrent.SynchronousQueue;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.client.HedwigClient;
-import org.apache.hedwig.client.api.Publisher;
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions;
-import org.apache.hedwig.server.HedwigRegionTestBase;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.server.integration.TestHedwigHub.TestCallback;
-import org.apache.hedwig.server.integration.TestHedwigHub.TestMessageHandler;
-import org.apache.hedwig.util.HedwigSocketAddress;
-
-import static org.junit.Assert.*;
-
-@RunWith(Parameterized.class)
-public class TestHedwigRegion extends HedwigRegionTestBase {
-
-    // SynchronousQueues to verify async calls
-    private final SynchronousQueue<Boolean> queue = new SynchronousQueue<Boolean>();
-    private final SynchronousQueue<Boolean> consumeQueue = new SynchronousQueue<Boolean>();
-
-    private static final int TEST_RETRY_REMOTE_SUBSCRIBE_INTERVAL_VALUE = 3000;
-
-    protected class NewRegionServerConfiguration extends RegionServerConfiguration {
-
-        public NewRegionServerConfiguration(int serverPort, int sslServerPort,
-                String regionName) {
-            super(serverPort, sslServerPort, regionName);
-        }
-
-        @Override
-        public int getRetryRemoteSubscribeThreadRunInterval() {
-            return TEST_RETRY_REMOTE_SUBSCRIBE_INTERVAL_VALUE;
-        }
-
-    }
-
-    protected class NewRegionClientConfiguration extends ClientConfiguration {
-        @Override
-        public boolean isSubscriptionChannelSharingEnabled() {
-            return isSubscriptionChannelSharingEnabled;
-        }
-        @Override
-        public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
-            return regionHubAddresses.get(0).get(0);
-        }
-    }
-
-    protected ServerConfiguration getServerConfiguration(int serverPort, int sslServerPort, String regionName) {
-        return new NewRegionServerConfiguration(serverPort, sslServerPort, regionName);
-    }
-
-    protected ClientConfiguration getRegionClientConfiguration() {
-        return new NewRegionClientConfiguration();
-    }
-
-    @Parameters
-    public static Collection<Object[]> configs() {
-        return Arrays.asList(new Object[][] { { false }, { true } });
-    }
-
-    protected boolean isSubscriptionChannelSharingEnabled;
-
-    public TestHedwigRegion(boolean isSubscriptionChannelSharingEnabled) {
-        this.isSubscriptionChannelSharingEnabled = isSubscriptionChannelSharingEnabled;
-    }
-
-    @Override
-    @Before
-    public void setUp() throws Exception {
-        numRegions = 3;
-        numServersPerRegion = 4;
-        super.setUp();
-    }
-
-    @Test(timeout=60000)
-    public void testMultiRegionSubscribeAndConsume() throws Exception {
-        int batchSize = 10;
-        // Subscribe to topics for clients in all regions
-        for (HedwigClient client : regionClientsMap.values()) {
-            for (int i = 0; i < batchSize; i++) {
-                SubscriptionOptions opts = SubscriptionOptions.newBuilder()
-                    .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).build();
-
-                client.getSubscriber().asyncSubscribe(ByteString.copyFromUtf8("Topic" + i),
-                        ByteString.copyFromUtf8("LocalSubscriber"), opts, new TestCallback(queue), null);
-                assertTrue(queue.take());
-            }
-        }
-
-        // Start delivery for the local subscribers in all regions
-        for (HedwigClient client : regionClientsMap.values()) {
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().startDelivery(ByteString.copyFromUtf8("Topic" + i),
-                                                     ByteString.copyFromUtf8("LocalSubscriber"), new TestMessageHandler(consumeQueue));
-            }
-        }
-
-        // Now start publishing messages for the subscribed topics in one of the
-        // regions and verify that it gets delivered and consumed in all of the
-        // other ones.
-        Publisher publisher = regionClientsMap.values().iterator().next().getPublisher();
-        for (int i = 0; i < batchSize; i++) {
-            publisher.asyncPublish(ByteString.copyFromUtf8("Topic" + i), Message.newBuilder().setBody(
-                                       ByteString.copyFromUtf8("Message" + i)).build(), new TestCallback(queue), null);
-            assertTrue(queue.take());
-        }
-        // Make sure each region consumes the same set of published messages.
-        for (int i = 0; i < regionClientsMap.size(); i++) {
-            for (int j = 0; j < batchSize; j++) {
-                assertTrue(consumeQueue.take());
-            }
-        }
-    }
-
-    /**
-     * Test region shuts down when first subscription.
-     *
-     * @throws Exception
-     */
-    @Test(timeout=60000)
-    public void testSubscribeAndConsumeWhenARegionDown() throws Exception {
-        int batchSize = 10;
-
-        SubscriptionOptions opts = SubscriptionOptions.newBuilder()
-            .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).build();
-
-        // first shut down a region
-        Random r = new Random();
-        int regionId = r.nextInt(numRegions);
-        stopRegion(regionId);
-        // subscribe to topics when a region shuts down
-        for (HedwigClient client : regionClientsMap.values()) {
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().asyncSubscribe(ByteString.copyFromUtf8("Topic" + i),
-                                                      ByteString.copyFromUtf8("LocalSubscriber"),
-                                                      opts, new TestCallback(queue), null);
-                assertFalse(queue.take());
-            }
-        }
-
-        // start region gain
-        startRegion(regionId);
-
-        // sub it again
-        for (Map.Entry<String, HedwigClient> entry : regionClientsMap.entrySet()) {
-            HedwigClient client = entry.getValue();
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().asyncSubscribe(ByteString.copyFromUtf8("Topic" + i),
-                                                      ByteString.copyFromUtf8("LocalSubscriber"),
-                                                      opts, new TestCallback(queue), null);
-                assertTrue(queue.take());
-            }
-        }
-
-        // Start delivery for local subscribers in all regions
-        for (Map.Entry<String, HedwigClient> entry : regionClientsMap.entrySet()) {
-            HedwigClient client = entry.getValue();
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().startDelivery(ByteString.copyFromUtf8("Topic" + i),
-                                                     ByteString.copyFromUtf8("LocalSubscriber"), new TestMessageHandler(consumeQueue));
-            }
-        }
-
-        // Now start publishing messages for the subscribed topics in one of the
-        // regions and verify that it gets delivered and consumed in all of the
-        // other ones.
-        int rid = r.nextInt(numRegions);
-        String regionName = REGION_PREFIX + rid;
-        Publisher publisher = regionClientsMap.get(regionName).getPublisher();
-        for (int i = 0; i < batchSize; i++) {
-            publisher.asyncPublish(ByteString.copyFromUtf8("Topic" + i), Message.newBuilder().setBody(
-                                   ByteString.copyFromUtf8(regionName + "-Message" + i)).build(), new TestCallback(queue), null);
-            assertTrue(queue.take());
-        }
-        // Make sure each region consumes the same set of published messages.
-        for (int i = 0; i < regionClientsMap.size(); i++) {
-            for (int j = 0; j < batchSize; j++) {
-                assertTrue(consumeQueue.take());
-            }
-        }
-    }
-
-    /**
-     * Test region shuts down when attaching existing subscriptions.
-     *
-     * @throws Exception
-     */
-    @Test(timeout=60000)
-    public void testAttachExistingSubscriptionsWhenARegionDown() throws Exception {
-        int batchSize = 10;
-
-        SubscriptionOptions opts = SubscriptionOptions.newBuilder()
-            .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).build();
-
-        // sub it remotely to make subscriptions existed
-        for (Map.Entry<String, HedwigClient> entry : regionClientsMap.entrySet()) {
-            HedwigClient client = entry.getValue();
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().asyncSubscribe(ByteString.copyFromUtf8("Topic" + i),
-                                                      ByteString.copyFromUtf8("LocalSubscriber"),
-                                                      opts, new TestCallback(queue), null);
-                assertTrue(queue.take());
-            }
-        }
-
-        // stop regions
-        for (int i=0; i<numRegions; i++) {
-            stopRegion(i);
-        }
-        // start regions again
-        for (int i=0; i<numRegions; i++) {
-            startRegion(i);
-        }
-
-        // first shut down a region
-        Random r = new Random();
-        int regionId = r.nextInt(numRegions);
-        stopRegion(regionId);
-        // subscribe to topics when a region shuts down
-        // it should succeed since the subscriptions existed before
-        for (HedwigClient client : regionClientsMap.values()) {
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().asyncSubscribe(ByteString.copyFromUtf8("Topic" + i),
-                                                      ByteString.copyFromUtf8("LocalSubscriber"),
-                                                      opts, new TestCallback(queue), null);
-                assertTrue(queue.take());
-            }
-        }
-
-        // Start delivery for local subscribers in all regions
-        for (Map.Entry<String, HedwigClient> entry : regionClientsMap.entrySet()) {
-            HedwigClient client = entry.getValue();
-            for (int i = 0; i < batchSize; i++) {
-                client.getSubscriber().startDelivery(ByteString.copyFromUtf8("Topic" + i),
-                                                     ByteString.copyFromUtf8("LocalSubscriber"), new TestMessageHandler(consumeQueue));
-            }
-        }
-
-        // start region again
-        startRegion(regionId);
-        // wait for retry
-        Thread.sleep(3 * TEST_RETRY_REMOTE_SUBSCRIBE_INTERVAL_VALUE);
-
-        String regionName = REGION_PREFIX + regionId;
-        HedwigClient client = regionClientsMap.get(regionName);
-        for (int i = 0; i < batchSize; i++) {
-            client.getSubscriber().asyncSubscribe(ByteString.copyFromUtf8("Topic" + i),
-                                                  ByteString.copyFromUtf8("LocalSubscriber"),
-                                                  opts, new TestCallback(queue), null);
-            assertTrue(queue.take());
-            client.getSubscriber().startDelivery(ByteString.copyFromUtf8("Topic" + i),
-                    ByteString.copyFromUtf8("LocalSubscriber"), new TestMessageHandler(consumeQueue));
-        }
-
-        // Now start publishing messages for the subscribed topics in one of the
-        // regions and verify that it gets delivered and consumed in all of the
-        // other ones.
-        Publisher publisher = client.getPublisher();
-        for (int i = 0; i < batchSize; i++) {
-            publisher.asyncPublish(ByteString.copyFromUtf8("Topic" + i), Message.newBuilder().setBody(
-                                   ByteString.copyFromUtf8(regionName + "-Message" + i)).build(), new TestCallback(queue), null);
-            assertTrue(queue.take());
-        }
-        // Make sure each region consumes the same set of published messages.
-        for (int i = 0; i < regionClientsMap.size(); i++) {
-            for (int j = 0; j < batchSize; j++) {
-                assertTrue(consumeQueue.take());
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/meta/MetadataManagerFactoryTestCase.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/MetadataManagerFactoryTestCase.java b/hedwig-server/src/test/java/org/apache/hedwig/server/meta/MetadataManagerFactoryTestCase.java
deleted file mode 100644
index 644500d..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/MetadataManagerFactoryTestCase.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- *
- * 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.hedwig.server.meta;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-import org.apache.bookkeeper.metastore.InMemoryMetaStore;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.server.meta.MetadataManagerFactory;
-import org.apache.hedwig.server.meta.ZkMetadataManagerFactory;
-import org.apache.hedwig.zookeeper.ZooKeeperTestBase;
-
-import org.junit.After;
-import org.junit.Before;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@RunWith(Parameterized.class)
-public abstract class MetadataManagerFactoryTestCase extends ZooKeeperTestBase {
-    private static final Logger LOG = LoggerFactory.getLogger(MetadataManagerFactoryTestCase.class);
-
-    protected MetadataManagerFactory metadataManagerFactory;
-    protected ServerConfiguration conf;
-
-    public MetadataManagerFactoryTestCase(String metadataManagerFactoryCls) {
-        super();
-        conf = new ServerConfiguration();
-        conf.setMetadataManagerFactoryName(metadataManagerFactoryCls);
-        conf.getConf().setProperty("metastore_impl_class", InMemoryMetaStore.class.getName());
-        InMemoryMetaStore.reset();
-    }
-
-    @Parameters
-    public static Collection<Object[]> configs() {
-        return Arrays.asList(new Object[][] {
-            { ZkMetadataManagerFactory.class.getName() },
-            { MsMetadataManagerFactory.class.getName() },
-        });
-    }
-
-    @Before
-    @Override
-    public void setUp() throws Exception {
-        super.setUp();
-        metadataManagerFactory = MetadataManagerFactory.newMetadataManagerFactory(conf, zk);
-    }
-
-    @After
-    @Override
-    public void tearDown() throws Exception {
-        metadataManagerFactory.shutdown();
-        super.tearDown();
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestFactoryLayout.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestFactoryLayout.java b/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestFactoryLayout.java
deleted file mode 100644
index 7e15135..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestFactoryLayout.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- *
- * 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.hedwig.server.meta;
-
-import java.io.IOException;
-
-import org.apache.hedwig.protocol.PubSubProtocol.ManagerMeta;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.zookeeper.ZooKeeperTestBase;
-import org.apache.hedwig.zookeeper.ZkUtils;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.ZooDefs.Ids;
-
-import org.junit.Test;
-import org.junit.Assert;
-
-public class TestFactoryLayout extends ZooKeeperTestBase {
-
-    @Test(timeout=60000)
-    public void testFactoryLayout() throws Exception {
-        ServerConfiguration conf = new ServerConfiguration();
-        conf.setMetadataManagerFactoryName(
-            "org.apache.hedwig.server.meta.ZkMetadataManager");
-
-        FactoryLayout layout = FactoryLayout.readLayout(zk, conf);
-        Assert.assertTrue("Layout should be null", layout == null);
-
-        String testName = "foobar";
-        int testVersion = 0xdeadbeef;
-        // use layout defined in configuration also create it in zookeeper
-        writeFactoryLayout(conf, testName, testVersion);
-
-        layout = FactoryLayout.readLayout(zk, conf);
-        Assert.assertEquals(testName, layout.getManagerMeta().getManagerImpl());
-        Assert.assertEquals(testVersion, layout.getManagerMeta().getManagerVersion());
-    }
-
-    private void writeFactoryLayout(ServerConfiguration conf, String managerCls,
-                                    int managerVersion)
-        throws Exception {
-        ManagerMeta managerMeta = ManagerMeta.newBuilder()
-                                             .setManagerImpl(managerCls)
-                                             .setManagerVersion(managerVersion)
-                                             .build();
-        FactoryLayout layout = new FactoryLayout(managerMeta);
-        layout.store(zk, conf);
-    }
-
-    @Test(timeout=60000)
-    public void testCorruptedFactoryLayout() throws Exception {
-        ServerConfiguration conf = new ServerConfiguration();
-        StringBuilder msb = new StringBuilder();
-        String factoryLayoutPath = FactoryLayout.getFactoryLayoutPath(msb, conf);
-        // write corrupted manager layout
-        ZkUtils.createFullPathOptimistic(zk, factoryLayoutPath, "BadLayout".getBytes(),
-                                         Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-
-        try {
-            FactoryLayout.readLayout(zk, conf);
-            Assert.fail("Shouldn't reach here!");
-        } catch (IOException ie) {
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java b/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java
deleted file mode 100644
index 7e395e9..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java
+++ /dev/null
@@ -1,365 +0,0 @@
-/*
- *
- * 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.hedwig.server.meta;
-
-import java.util.Map;
-
-import com.google.protobuf.ByteString;
-
-import org.apache.bookkeeper.versioning.Version;
-import org.apache.bookkeeper.versioning.Versioned;
-import org.apache.hedwig.StubCallback;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.protocol.PubSubProtocol.LedgerRanges;
-import org.apache.hedwig.protocol.PubSubProtocol.LedgerRange;
-import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
-import org.apache.hedwig.server.topics.HubInfo;
-import org.apache.hedwig.util.Callback;
-import org.apache.hedwig.util.Either;
-import org.apache.hedwig.util.HedwigSocketAddress;
-
-import org.junit.Test;
-import org.junit.Assert;
-
-public class TestMetadataManager extends MetadataManagerFactoryTestCase {
-
-    public TestMetadataManager(String metadataManagerFactoryCls) {
-        super(metadataManagerFactoryCls);
-    }
-
-    @Test(timeout=60000)
-    public void testOwnerInfo() throws Exception {
-        TopicOwnershipManager toManager = metadataManagerFactory.newTopicOwnershipManager();
-
-        ByteString topic = ByteString.copyFromUtf8("testOwnerInfo");
-        StubCallback<Versioned<HubInfo>> readCallback = new StubCallback<Versioned<HubInfo>>();
-        StubCallback<Version> writeCallback = new StubCallback<Version>();
-        StubCallback<Void> deleteCallback = new StubCallback<Void>();
-
-        Either<Version, PubSubException> res;
-        HubInfo owner = new HubInfo(new HedwigSocketAddress("127.0.0.1", 8008), 999);
-
-        // Write non-existed owner info
-        toManager.writeOwnerInfo(topic, owner, Version.NEW, writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertEquals(null, res.right());
-        Version v1 = res.left();
-
-        // read owner info
-        toManager.readOwnerInfo(topic, readCallback, null);
-        Versioned<HubInfo> hubInfo = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v1.compare(hubInfo.getVersion()));
-        Assert.assertEquals(owner, hubInfo.getValue());
-
-        HubInfo newOwner = new HubInfo(new HedwigSocketAddress("127.0.0.1", 8008), 1000);
-
-        // write exsited owner info with null version
-        toManager.writeOwnerInfo(topic, newOwner, Version.NEW, writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertNotNull(res.right());
-        Assert.assertTrue(res.right() instanceof PubSubException.TopicOwnerInfoExistsException);
-
-        // write existed owner info with right version
-        toManager.writeOwnerInfo(topic, newOwner, v1, writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertEquals(null, res.right());
-        Version v2 = res.left();
-        Assert.assertEquals(Version.Occurred.AFTER, v2.compare(v1));
-
-        // read owner info
-        toManager.readOwnerInfo(topic, readCallback, null);
-        hubInfo = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(hubInfo.getVersion()));
-        Assert.assertEquals(newOwner, hubInfo.getValue());
-
-        HubInfo newOwner2 = new HubInfo(new HedwigSocketAddress("127.0.0.1", 8008), 1001);
-
-        // write existed owner info with bad version
-        toManager.writeOwnerInfo(topic, newOwner2, v1,
-                                 writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertNotNull(res.right());
-        Assert.assertTrue(res.right() instanceof PubSubException.BadVersionException);
-
-        // read owner info
-        toManager.readOwnerInfo(topic, readCallback, null);
-        hubInfo = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(hubInfo.getVersion()));
-        Assert.assertEquals(newOwner, hubInfo.getValue());
-
-        // delete existed owner info with bad version
-        toManager.deleteOwnerInfo(topic, v1, deleteCallback, null);
-        Assert.assertTrue(deleteCallback.queue.take().right() instanceof
-                          PubSubException.BadVersionException);
-
-        // read owner info
-        toManager.readOwnerInfo(topic, readCallback, null);
-        hubInfo = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(hubInfo.getVersion()));
-
-        // delete existed owner info with right version
-        toManager.deleteOwnerInfo(topic, v2, deleteCallback, null);
-        Assert.assertEquals(null, deleteCallback.queue.take().right());
-
-        // Empty owner info
-        toManager.readOwnerInfo(topic, readCallback, null);
-        Assert.assertEquals(null, readCallback.queue.take().left());
-
-        // delete non-existed owner info
-        toManager.deleteOwnerInfo(topic, Version.ANY, deleteCallback, null);
-        Assert.assertTrue(deleteCallback.queue.take().right() instanceof
-                          PubSubException.NoTopicOwnerInfoException);
-
-        toManager.close();
-    }
-
-    @Test(timeout=60000)
-    public void testPersistenceInfo() throws Exception {
-        TopicPersistenceManager tpManager = metadataManagerFactory.newTopicPersistenceManager();
-
-        ByteString topic = ByteString.copyFromUtf8("testPersistenceInfo");
-        StubCallback<Versioned<LedgerRanges>> readCallback = new StubCallback<Versioned<LedgerRanges>>();
-        StubCallback<Version> writeCallback = new StubCallback<Version>();
-        StubCallback<Void> deleteCallback = new StubCallback<Void>();
-
-        // Write non-existed persistence info
-        tpManager.writeTopicPersistenceInfo(topic, LedgerRanges.getDefaultInstance(),
-                                            Version.NEW, writeCallback, null);
-        Either<Version, PubSubException> res = writeCallback.queue.take();
-        Assert.assertEquals(null, res.right());
-        Version v1 = res.left();
-
-        // read persistence info
-        tpManager.readTopicPersistenceInfo(topic, readCallback, null);
-        Versioned<LedgerRanges> ranges = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v1.compare(ranges.getVersion()));
-        Assert.assertEquals(LedgerRanges.getDefaultInstance(), ranges.getValue());
-
-        LedgerRange lastRange = LedgerRange.newBuilder().setLedgerId(1).build();
-        LedgerRanges.Builder builder = LedgerRanges.newBuilder();
-        builder.addRanges(lastRange);
-        LedgerRanges newRanges = builder.build();
-
-        // write existed persistence info with null version
-        tpManager.writeTopicPersistenceInfo(topic, newRanges, Version.NEW,
-                                            writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertNotNull(res.right());
-        Assert.assertTrue(res.right() instanceof PubSubException.TopicPersistenceInfoExistsException);
-
-        // write existed persistence info with right version
-        tpManager.writeTopicPersistenceInfo(topic, newRanges, v1,
-                                            writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertEquals(null, res.right());
-        Version v2 = res.left();
-        Assert.assertEquals(Version.Occurred.AFTER, v2.compare(v1));
-
-        // read persistence info
-        tpManager.readTopicPersistenceInfo(topic, readCallback, null);
-        ranges = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(ranges.getVersion()));
-        Assert.assertEquals(newRanges, ranges.getValue());
-
-        lastRange = LedgerRange.newBuilder().setLedgerId(2).build();
-        builder = LedgerRanges.newBuilder();
-        builder.addRanges(lastRange);
-        LedgerRanges newRanges2 = builder.build();
-
-        // write existed persistence info with bad version
-        tpManager.writeTopicPersistenceInfo(topic, newRanges2, v1,
-                                            writeCallback, null);
-        res = writeCallback.queue.take();
-        Assert.assertNotNull(res.right());
-        Assert.assertTrue(res.right() instanceof PubSubException.BadVersionException);
-
-        // read persistence info
-        tpManager.readTopicPersistenceInfo(topic, readCallback, null);
-        ranges = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(ranges.getVersion()));
-        Assert.assertEquals(newRanges, ranges.getValue());
-
-        // delete with bad version
-        tpManager.deleteTopicPersistenceInfo(topic, v1, deleteCallback, null);
-        Assert.assertTrue(deleteCallback.queue.take().right() instanceof
-                          PubSubException.BadVersionException);
-
-        // read persistence info
-        tpManager.readTopicPersistenceInfo(topic, readCallback, null);
-        ranges = readCallback.queue.take().left();
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(ranges.getVersion()));
-        Assert.assertEquals(newRanges, ranges.getValue());
-
-        // delete existed persistence info with right version
-        tpManager.deleteTopicPersistenceInfo(topic, v2, deleteCallback, null);
-        Assert.assertEquals(null, deleteCallback.queue.take().right());
-
-        // read empty persistence info
-        tpManager.readTopicPersistenceInfo(topic, readCallback, null);
-        Assert.assertEquals(null, readCallback.queue.take().left());
-
-        // delete non-existed persistence info
-        tpManager.deleteTopicPersistenceInfo(topic, Version.ANY, deleteCallback, null);
-        Assert.assertTrue(deleteCallback.queue.take().right() instanceof
-                          PubSubException.NoTopicPersistenceInfoException);
-
-        tpManager.close();
-    }
-
-    @Test(timeout=60000)
-    public void testSubscriptionData() throws Exception {
-        SubscriptionDataManager subManager = metadataManagerFactory.newSubscriptionDataManager();
-
-        ByteString topic = ByteString.copyFromUtf8("testSubscriptionData");
-        ByteString subid = ByteString.copyFromUtf8("mysub");
-
-        final StubCallback<Version> callback = new StubCallback<Version>();
-        StubCallback<Versioned<SubscriptionData>> readCallback = new StubCallback<Versioned<SubscriptionData>>();
-        StubCallback<Map<ByteString, Versioned<SubscriptionData>>> subsCallback
-            = new StubCallback<Map<ByteString, Versioned<SubscriptionData>>>();
-
-        subManager.readSubscriptionData(topic, subid, readCallback, null);
-        Either<Versioned<SubscriptionData>, PubSubException> readRes = readCallback.queue.take();
-        Assert.assertEquals("Found inconsistent subscription state", null, readRes.left());
-        Assert.assertEquals("Should not fail with PubSubException", null, readRes.right());
-
-        // read non-existed subscription state
-        subManager.readSubscriptions(topic, subsCallback, null);
-        Either<Map<ByteString, Versioned<SubscriptionData>>, PubSubException> res = subsCallback.queue.take();
-        Assert.assertEquals("Found more than 0 subscribers", 0, res.left().size());
-        Assert.assertEquals("Should not fail with PubSubException", null, res.right());
-
-        // update non-existed subscription state
-        if (subManager.isPartialUpdateSupported()) {
-            subManager.updateSubscriptionData(topic, subid, 
-                    SubscriptionData.getDefaultInstance(), Version.ANY, callback, null);
-        } else {
-            subManager.replaceSubscriptionData(topic, subid, 
-                    SubscriptionData.getDefaultInstance(), Version.ANY, callback, null);
-        }
-        Assert.assertTrue("Should fail to update a non-existed subscriber with PubSubException",
-                          callback.queue.take().right() instanceof PubSubException.NoSubscriptionStateException);
-
-        Callback<Void> voidCallback = new Callback<Void>() {
-            @Override
-            public void operationFinished(Object ctx, Void resultOfOperation) {
-                callback.operationFinished(ctx, null);
-            }
-            @Override
-            public void operationFailed(Object ctx, PubSubException exception) {
-                callback.operationFailed(ctx, exception);
-            }
-        }; 
-        
-        // delete non-existed subscription state
-        subManager.deleteSubscriptionData(topic, subid, Version.ANY, voidCallback, null);
-        Assert.assertTrue("Should fail to delete a non-existed subscriber with PubSubException",
-                          callback.queue.take().right() instanceof PubSubException.NoSubscriptionStateException);
-
-        long seqId = 10;
-        MessageSeqId.Builder builder = MessageSeqId.newBuilder();
-        builder.setLocalComponent(seqId);
-        MessageSeqId msgId = builder.build();
-
-        SubscriptionState.Builder stateBuilder = SubscriptionState.newBuilder(SubscriptionState.getDefaultInstance()).setMsgId(msgId);
-        SubscriptionData data = SubscriptionData.newBuilder().setState(stateBuilder).build();
-
-        // create a subscription state
-        subManager.createSubscriptionData(topic, subid, data, callback, null);
-        Either<Version, PubSubException> cbResult = callback.queue.take();
-        Version v1 = cbResult.left();
-        Assert.assertEquals("Should not fail with PubSubException",
-                            null, cbResult.right());
-
-        // read subscriptions
-        subManager.readSubscriptions(topic, subsCallback, null);
-        res = subsCallback.queue.take();
-        Assert.assertEquals("Should find just 1 subscriber", 1, res.left().size());
-        Assert.assertEquals("Should not fail with PubSubException", null, res.right());
-        Versioned<SubscriptionData> versionedSubData = res.left().get(subid);
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v1.compare(versionedSubData.getVersion()));
-        SubscriptionData imss = versionedSubData.getValue();
-        Assert.assertEquals("Found inconsistent subscription state",
-                            data, imss);
-        Assert.assertEquals("Found inconsistent last consumed seq id",
-                            seqId, imss.getState().getMsgId().getLocalComponent());
-
-        // move consume seq id
-        seqId = 99;
-        builder = MessageSeqId.newBuilder();
-        builder.setLocalComponent(seqId);
-        msgId = builder.build();
-
-        stateBuilder = SubscriptionState.newBuilder(data.getState()).setMsgId(msgId);
-        data = SubscriptionData.newBuilder().setState(stateBuilder).build();
-        
-        // update subscription state
-        if (subManager.isPartialUpdateSupported()) {
-            subManager.updateSubscriptionData(topic, subid, data, versionedSubData.getVersion(), callback, null);
-        } else {
-            subManager.replaceSubscriptionData(topic, subid, data, versionedSubData.getVersion(), callback, null);
-        }
-        cbResult = callback.queue.take();
-        Assert.assertEquals("Fail to update a subscription state", null, cbResult.right());
-        Version v2 = cbResult.left();
-        // read subscription state
-        subManager.readSubscriptionData(topic, subid, readCallback, null);
-        Assert.assertEquals("Found inconsistent subscription state",
-                            data, readCallback.queue.take().left().getValue());
-        
-        // read subscriptions again
-        subManager.readSubscriptions(topic, subsCallback, null);
-        res = subsCallback.queue.take();
-        Assert.assertEquals("Should find just 1 subscriber", 1, res.left().size());
-        Assert.assertEquals("Should not fail with PubSubException", null, res.right());
-        versionedSubData = res.left().get(subid);
-        Assert.assertEquals(Version.Occurred.CONCURRENTLY, v2.compare(versionedSubData.getVersion()));
-        imss = res.left().get(subid).getValue();
-        Assert.assertEquals("Found inconsistent subscription state",
-                            data, imss);
-        Assert.assertEquals("Found inconsistent last consumed seq id",
-                            seqId, imss.getState().getMsgId().getLocalComponent());
-
-        // update or replace subscription data with bad version
-        if (subManager.isPartialUpdateSupported()) {
-            subManager.updateSubscriptionData(topic, subid, data, v1, callback, null);
-        } else {
-            subManager.replaceSubscriptionData(topic, subid, data, v1, callback, null);
-        }
-        Assert.assertTrue(callback.queue.take().right() instanceof PubSubException.BadVersionException);
-        
-        // delete with bad version
-        subManager.deleteSubscriptionData(topic, subid, v1, voidCallback, null);
-        Assert.assertTrue(callback.queue.take().right() instanceof PubSubException.BadVersionException);
-        subManager.deleteSubscriptionData(topic, subid, res.left().get(subid).getVersion(), voidCallback, null);
-        Assert.assertEquals("Fail to delete an existed subscriber", null, callback.queue.take().right());
-
-        // read subscription states again
-        subManager.readSubscriptions(topic, subsCallback, null);
-        res = subsCallback.queue.take();
-        Assert.assertEquals("Found more than 0 subscribers", 0, res.left().size());
-        Assert.assertEquals("Should not fail with PubSubException", null, res.right());
-
-        subManager.close();
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManagerFactory.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManagerFactory.java b/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManagerFactory.java
deleted file mode 100644
index 8b9016a..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManagerFactory.java
+++ /dev/null
@@ -1,299 +0,0 @@
-/*
- *
- * 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.hedwig.server.meta;
-
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooKeeper;
-import java.io.IOException;
-import java.util.concurrent.CyclicBarrier;
-import java.util.concurrent.CountDownLatch;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.protocol.PubSubProtocol.ManagerMeta;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.zookeeper.ZooKeeperTestBase;
-import org.junit.Test;
-import org.junit.Assert;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TestMetadataManagerFactory extends ZooKeeperTestBase {
-    private static final Logger LOG = LoggerFactory.getLogger(TestMetadataManagerFactory.class);
-
-    static class TestServerConfiguration extends ServerConfiguration {
-        String hedwigPrefix = "/hedwig";
-
-        @Override
-        public String getZkPrefix() {
-            return hedwigPrefix;
-        }
-
-        public void setZkPrefix(String prefix) {
-            this.hedwigPrefix = prefix;
-        }
-    }
-
-    static class DummyMetadataManagerFactory extends MetadataManagerFactory {
-        static int VERSION = 10;
-
-        public int getCurrentVersion() { return VERSION; }
-
-
-        public MetadataManagerFactory initialize(ServerConfiguration cfg,
-                                                 ZooKeeper zk,
-                                                 int version)
-        throws IOException {
-            if (version != VERSION) {
-                throw new IOException("unmatched manager version");
-            }
-            // do nothing
-            return this;
-        }
-
-        public void shutdown() {}
-
-        public Iterator<ByteString> getTopics() {
-            return null;
-        }
-
-        public TopicPersistenceManager newTopicPersistenceManager() {
-            return null;
-        }
-
-        public SubscriptionDataManager newSubscriptionDataManager() {
-            return null;
-        }
-
-        public TopicOwnershipManager newTopicOwnershipManager() {
-            return null;
-        }
-
-        public void format(ServerConfiguration cfg, ZooKeeper zk) throws IOException {
-            // do nothing
-        }
-    }
-
-    private void writeFactoryLayout(ServerConfiguration conf,
-                                    String factoryCls,
-                                    int factoryVersion)
-        throws Exception {
-        ManagerMeta meta = ManagerMeta.newBuilder()
-                                      .setManagerImpl(factoryCls)
-                                      .setManagerVersion(factoryVersion).build();
-        new FactoryLayout(meta).store(zk, conf);
-    }
-
-    /**
-     * Test bad server configuration
-     */
-    @Test(timeout=60000)
-    public void testBadConf() throws Exception {
-        TestServerConfiguration conf = new TestServerConfiguration();
-
-        String root0 = "/goodconf";
-        conf.setZkPrefix(root0);
-
-        MetadataManagerFactory m =
-            MetadataManagerFactory.newMetadataManagerFactory(conf, zk);
-        Assert.assertTrue("MetadataManagerFactory is unexpected type",
-                          (m instanceof ZkMetadataManagerFactory));
-
-        // mismatching conf
-        conf.setMetadataManagerFactoryName(DummyMetadataManagerFactory.class.getName());
-        try {
-            MetadataManagerFactory.newMetadataManagerFactory(conf, zk);
-            Assert.fail("Shouldn't reach here");
-        } catch (Exception e) {
-            Assert.assertTrue("Invalid exception",
-                              e.getMessage().contains("does not match existing factory"));
-        }
-
-        // invalid metadata manager
-        String root1 = "/badconf1";
-        conf.setZkPrefix(root1);
-        conf.setMetadataManagerFactoryName("DoesNotExist");
-        try {
-            MetadataManagerFactory.newMetadataManagerFactory(conf, zk);
-            Assert.fail("Shouldn't reach here");
-        } catch (Exception e) {
-            Assert.assertTrue("Invalid exception",
-                              e.getMessage().contains("Failed to get metadata manager factory class from configuration"));
-        }
-    }
-
-    /**
-     * Test bad zk configuration
-     */
-    @Test(timeout=60000)
-    public void testBadZkContents() throws Exception {
-        TestServerConfiguration conf = new TestServerConfiguration();
-
-        // bad type in zookeeper
-        String root0 = "/badzk0";
-        conf.setZkPrefix(root0);
-
-        writeFactoryLayout(conf, "DoesNotExist", 0xdeadbeef);
-        try {
-            MetadataManagerFactory.newMetadataManagerFactory(conf, zk);
-            Assert.fail("Shouldn't reach here");
-        } catch (Exception e) {
-            Assert.assertTrue("Invalid exception",
-                              e.getMessage().contains("No class found to instantiate metadata manager factory"));
-        }
-
-        // bad version in zookeeper
-        String root1 = "/badzk1";
-        conf.setZkPrefix(root1);
-
-        writeFactoryLayout(conf, ZkMetadataManagerFactory.class.getName(), 0xdeadbeef);
-        try {
-            MetadataManagerFactory.newMetadataManagerFactory(conf, zk);
-            Assert.fail("Shouldn't reach here");
-        } catch (Exception e) {
-            Assert.assertTrue("Invalid exception",
-                              e.getMessage().contains("Incompatible ZkMetadataManagerFactory version"));
-        }
-    }
-
-    private class CreateMMThread extends Thread {
-        private boolean success = false;
-        private final String factoryCls;
-        private final String root;
-        private final CyclicBarrier barrier;
-        private ZooKeeper zkc;
-
-        CreateMMThread(String root, String factoryCls, CyclicBarrier barrier) throws Exception {
-            this.factoryCls = factoryCls;
-            this.barrier = barrier;
-            this.root = root;
-            final CountDownLatch latch = new CountDownLatch(1);
-            zkc = new ZooKeeper(hostPort, 10000, new Watcher() {
-                public void process(WatchedEvent event) {
-                    latch.countDown();
-                }
-            });
-            latch.await();
-        }
-
-        public void run() {
-            TestServerConfiguration conf = new TestServerConfiguration();
-            conf.setZkPrefix(root);
-            conf.setMetadataManagerFactoryName(factoryCls);
-
-            try {
-                barrier.await();
-                MetadataManagerFactory.newMetadataManagerFactory(conf, zkc);
-                success = true;
-            } catch (Exception e) {
-                LOG.error("Failed to create metadata manager factory", e);
-            }
-        }
-
-        public boolean isSuccessful() {
-            return success;
-        }
-
-        public void close() throws Exception {
-            zkc.close();
-        }
-    }
-
-    // test concurrent
-    @Test(timeout=60000)
-    public void testConcurrent1() throws Exception {
-        /// everyone creates the same
-        int numThreads = 50;
-
-        // bad version in zookeeper
-        String root0 = "/lmroot0";
-
-        CyclicBarrier barrier = new CyclicBarrier(numThreads+1);
-        List<CreateMMThread> threads = new ArrayList<CreateMMThread>(numThreads);
-        for (int i = 0; i < numThreads; i++) {
-            CreateMMThread t = new CreateMMThread(root0, ZkMetadataManagerFactory.class.getName(), barrier);
-            t.start();
-            threads.add(t);
-        }
-
-        barrier.await();
-
-        boolean success = true;
-        for (CreateMMThread t : threads) {
-            t.join();
-            t.close();
-            success = t.isSuccessful() && success;
-        }
-        Assert.assertTrue("Not all metadata manager factories created", success);
-    }
-
-    @Test(timeout=60000)
-    public void testConcurrent2() throws Exception {
-        /// odd create different
-        int numThreadsEach = 25;
-
-        // bad version in zookeeper
-        String root0 = "/lmroot0";
-
-        CyclicBarrier barrier = new CyclicBarrier(numThreadsEach*2+1);
-        List<CreateMMThread> threadsA = new ArrayList<CreateMMThread>(numThreadsEach);
-        for (int i = 0; i < numThreadsEach; i++) {
-            CreateMMThread t = new CreateMMThread(root0, ZkMetadataManagerFactory.class.getName(), barrier);
-            t.start();
-            threadsA.add(t);
-        }
-        List<CreateMMThread> threadsB = new ArrayList<CreateMMThread>(numThreadsEach);
-        for (int i = 0; i < numThreadsEach; i++) {
-            CreateMMThread t = new CreateMMThread(root0, DummyMetadataManagerFactory.class.getName(), barrier);
-            t.start();
-            threadsB.add(t);
-        }
-
-        barrier.await();
-
-        int numSuccess = 0;
-        int numFails = 0;
-        for (CreateMMThread t : threadsA) {
-            t.join();
-            t.close();
-            if (t.isSuccessful()) {
-                numSuccess++;
-            } else {
-                numFails++;
-            }
-        }
-
-        for (CreateMMThread t : threadsB) {
-            t.join();
-            t.close();
-            if (t.isSuccessful()) {
-                numSuccess++;
-            } else {
-                numFails++;
-            }
-        }
-        Assert.assertEquals("Incorrect number of successes", numThreadsEach, numSuccess);
-        Assert.assertEquals("Incorrect number of failures", numThreadsEach, numFails);
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java b/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java
deleted file mode 100644
index 241f45b..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestPubSubServer.java
+++ /dev/null
@@ -1,265 +0,0 @@
-/**
- * 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.hedwig.server.netty;
-
-import java.io.IOException;
-import java.lang.Thread.UncaughtExceptionHandler;
-import java.net.InetSocketAddress;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.SynchronousQueue;
-
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooKeeper;
-import org.junit.Test;
-
-import org.apache.bookkeeper.test.PortManager;
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.client.HedwigClient;
-import org.apache.hedwig.client.api.Publisher;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.server.PubSubServerStandAloneTestBase;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.server.topics.AbstractTopicManager;
-import org.apache.hedwig.server.topics.TopicManager;
-import org.apache.hedwig.server.LoggingExceptionHandler;
-import org.apache.hedwig.util.Callback;
-import org.apache.hedwig.util.HedwigSocketAddress;
-import org.apache.hedwig.zookeeper.SafeAsyncZKCallback;
-
-import static org.junit.Assert.*;
-
-public class TestPubSubServer extends PubSubServerStandAloneTestBase {
-
-    @Test(timeout=60000)
-    public void testSecondServer() throws Exception {
-        PubSubServer server1 = new PubSubServer(new StandAloneServerConfiguration() {
-                @Override
-                public int getServerPort() {
-                    return super.getServerPort() + 1;
-                }
-            }, new ClientConfiguration(), new LoggingExceptionHandler());
-        server1.start();
-        server1.shutdown();
-    }
-
-    class RecordingUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
-        SynchronousQueue<Throwable> queue;
-
-        public RecordingUncaughtExceptionHandler(SynchronousQueue<Throwable> queue) {
-            this.queue = queue;
-        }
-
-        @Override
-        public void uncaughtException(Thread t, Throwable e) {
-            queue.add(e);
-        }
-
-    }
-
-    private interface TopicManagerInstantiator {
-        public TopicManager instantiateTopicManager() throws IOException;
-    }
-
-    PubSubServer startServer(final UncaughtExceptionHandler uncaughtExceptionHandler, final int port,
-                             final TopicManagerInstantiator instantiator) throws Exception {
-        PubSubServer server = new PubSubServer(new StandAloneServerConfiguration() {
-            @Override
-            public int getServerPort() {
-                return port;
-            }
-
-        }, new ClientConfiguration(), uncaughtExceptionHandler) {
-
-            @Override
-            protected TopicManager instantiateTopicManager() throws IOException {
-                return instantiator.instantiateTopicManager();
-            }
-        };
-        server.start();
-        return server;
-
-    }
-
-    public void runPublishRequest(final int port) throws Exception {
-        Publisher publisher = new HedwigClient(new ClientConfiguration() {
-            @Override
-            public InetSocketAddress getDefaultServerHost() {
-                return new InetSocketAddress("localhost", port);
-            }
-        }).getPublisher();
-
-        publisher.asyncPublish(ByteString.copyFromUtf8("blah"), Message.newBuilder().setBody(
-        ByteString.copyFromUtf8("blah")).build(), new Callback<Void>() {
-            @Override
-            public void operationFailed(Object ctx, PubSubException exception) {
-                assertTrue(false);
-            }
-
-            @Override
-            public void operationFinished(Object ctx, Void resultOfOperation) {
-                assertTrue(false);
-            }
-
-        }, null);
-    }
-
-    @Test(timeout=60000)
-    public void testUncaughtExceptionInNettyThread() throws Exception {
-
-        SynchronousQueue<Throwable> queue = new SynchronousQueue<Throwable>();
-        RecordingUncaughtExceptionHandler uncaughtExceptionHandler = new RecordingUncaughtExceptionHandler(queue);
-        final int port = PortManager.nextFreePort();
-
-        PubSubServer server = startServer(uncaughtExceptionHandler, port, new TopicManagerInstantiator() {
-
-            @Override
-            public TopicManager instantiateTopicManager() throws IOException {
-                return new AbstractTopicManager(new ServerConfiguration(), Executors.newSingleThreadScheduledExecutor()) {
-                    @Override
-                    protected void realGetOwner(ByteString topic, boolean shouldClaim,
-                    Callback<HedwigSocketAddress> cb, Object ctx) {
-                        throw new RuntimeException("this exception should be uncaught");
-                    }
-
-                    @Override
-                    protected void postReleaseCleanup(ByteString topic, Callback<Void> cb, Object ctx) {
-                    }
-                };
-            }
-        });
-
-        runPublishRequest(port);
-        assertEquals(RuntimeException.class, queue.take().getClass());
-        server.shutdown();
-    }
-
-    @Test(timeout=60000)
-    public void testUncaughtExceptionInZKThread() throws Exception {
-
-        SynchronousQueue<Throwable> queue = new SynchronousQueue<Throwable>();
-        RecordingUncaughtExceptionHandler uncaughtExceptionHandler = new RecordingUncaughtExceptionHandler(queue);
-        final int port = PortManager.nextFreePort();
-        final String hostPort = "127.0.0.1:" + PortManager.nextFreePort();
-
-        PubSubServer server = startServer(uncaughtExceptionHandler, port, new TopicManagerInstantiator() {
-
-            @Override
-            public TopicManager instantiateTopicManager() throws IOException {
-                return new AbstractTopicManager(new ServerConfiguration(), Executors.newSingleThreadScheduledExecutor()) {
-
-                    @Override
-                    protected void realGetOwner(ByteString topic, boolean shouldClaim,
-                    Callback<HedwigSocketAddress> cb, Object ctx) {
-                        ZooKeeper zookeeper;
-                        try {
-                            zookeeper = new ZooKeeper(hostPort, 60000, new Watcher() {
-                                @Override
-                                public void process(WatchedEvent event) {
-                                    // TODO Auto-generated method stub
-
-                                }
-                            });
-                        } catch (IOException e) {
-                            throw new RuntimeException(e);
-                        }
-
-                        zookeeper.getData("/fake", false, new SafeAsyncZKCallback.DataCallback() {
-                            @Override
-                            public void safeProcessResult(int rc, String path, Object ctx, byte[] data,
-                            org.apache.zookeeper.data.Stat stat) {
-                                throw new RuntimeException("This should go to the uncaught exception handler");
-                            }
-
-                        }, null);
-                    }
-
-                    @Override
-                    protected void postReleaseCleanup(ByteString topic, Callback<Void> cb, Object ctx) {
-                    }
-                };
-            }
-        });
-
-        runPublishRequest(port);
-        assertEquals(RuntimeException.class, queue.take().getClass());
-        server.shutdown();
-    }
-
-    @Test(timeout=60000)
-    public void testInvalidServerConfiguration() throws Exception {
-        boolean success = false;
-        ServerConfiguration conf = new ServerConfiguration() {
-            @Override
-            public boolean isInterRegionSSLEnabled() {
-                return conf.getBoolean(INTER_REGION_SSL_ENABLED, true);
-            }
-
-            @Override
-            public List<String> getRegions() {
-                List<String> regionsList = new LinkedList<String>();
-                regionsList.add("regionHost1:4080:9876");
-                regionsList.add("regionHost2:4080");
-                regionsList.add("regionHost3:4080:9876");
-                return regionsList;
-            }
-        };
-        try {
-            conf.validate();
-        }
-        catch (ConfigurationException e) {
-            logger.error("Invalid configuration: ", e);
-            success = true;
-        }
-        assertTrue(success);
-    }
-
-    @Test(timeout=60000)
-    public void testValidServerConfiguration() throws Exception {
-        boolean success = true;
-        ServerConfiguration conf = new ServerConfiguration() {
-            @Override
-            public boolean isInterRegionSSLEnabled() {
-                return conf.getBoolean(INTER_REGION_SSL_ENABLED, true);
-            }
-
-            @Override
-            public List<String> getRegions() {
-                List<String> regionsList = new LinkedList<String>();
-                regionsList.add("regionHost1:4080:9876");
-                regionsList.add("regionHost2:4080:2938");
-                regionsList.add("regionHost3:4080:9876");
-                return regionsList;
-            }
-        };
-        try {
-            conf.validate();
-        }
-        catch (ConfigurationException e) {
-            logger.error("Invalid configuration: ", e);
-            success = false;
-        }
-        assertTrue(success);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestServerStats.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestServerStats.java b/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestServerStats.java
deleted file mode 100644
index 08f5ad8..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/netty/TestServerStats.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/**
- * 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.hedwig.server.netty;
-
-import static org.junit.Assert.assertEquals;
-
-import org.apache.hedwig.server.netty.ServerStats.OpStats;
-import org.junit.Test;
-
-/** Tests that Statistics updation in hedwig Server */
-public class TestServerStats {
-
-    /**
-     * Tests that updatLatency should not fail with
-     * ArrayIndexOutOfBoundException when latency time coming as negative.
-     */
-    @Test(timeout=60000)
-    public void testUpdateLatencyShouldNotFailWithAIOBEWithNegativeLatency()
-            throws Exception {
-        OpStats opStat = new OpStats();
-        opStat.updateLatency(-10);
-        assertEquals("Should not update any latency metrics", 0,
-                opStat.numSuccessOps);
-
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/netty/WriteRecordingChannel.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/netty/WriteRecordingChannel.java b/hedwig-server/src/test/java/org/apache/hedwig/server/netty/WriteRecordingChannel.java
deleted file mode 100644
index 91cf2fe..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/netty/WriteRecordingChannel.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/**
- * 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.hedwig.server.netty;
-
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.util.LinkedList;
-import java.util.List;
-
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelConfig;
-import org.jboss.netty.channel.ChannelFactory;
-import org.jboss.netty.channel.ChannelFuture;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.DefaultChannelFuture;
-import org.jboss.netty.channel.SucceededChannelFuture;
-
-public class WriteRecordingChannel implements Channel {
-
-    public boolean closed = false;
-    ChannelFuture closingFuture = new DefaultChannelFuture(this, false);
-    List<Object> messagesWritten = new LinkedList<Object>();
-
-    public List<Object> getMessagesWritten() {
-        return messagesWritten;
-    }
-
-    public void clearMessages() {
-        messagesWritten.clear();
-    }
-
-    @Override
-    public ChannelFuture bind(SocketAddress localAddress) {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFuture close() {
-        closed = true;
-        closingFuture.setSuccess();
-        return new SucceededChannelFuture(this);
-    }
-
-    @Override
-    public ChannelFuture connect(SocketAddress remoteAddress) {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFuture disconnect() {
-        return close();
-    }
-
-    @Override
-    public ChannelFuture getCloseFuture() {
-        return closingFuture;
-    }
-
-    @Override
-    public ChannelConfig getConfig() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFactory getFactory() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public Integer getId() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public int getInterestOps() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public SocketAddress getLocalAddress() {
-        return new InetSocketAddress("localhost", 1234);
-    }
-
-    @Override
-    public Channel getParent() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelPipeline getPipeline() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public SocketAddress getRemoteAddress() {
-        return new InetSocketAddress("www.yahoo.com", 80);
-    }
-
-    @Override
-    public boolean isBound() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public boolean isConnected() {
-        return closed == false;
-    }
-
-    @Override
-    public boolean isOpen() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public boolean isReadable() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public boolean isWritable() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFuture setInterestOps(int interestOps) {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFuture setReadable(boolean readable) {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFuture unbind() {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public ChannelFuture write(Object message) {
-        messagesWritten.add(message);
-        return new SucceededChannelFuture(this);
-    }
-
-    @Override
-    public ChannelFuture write(Object message, SocketAddress remoteAddress) {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public int compareTo(Channel o) {
-        throw new RuntimeException("Not intended");
-    }
-
-    @Override
-    public void setAttachment(Object attachment) {}
-
-    @Override
-    public Object getAttachment() { throw new RuntimeException("Not intended"); }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java b/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java
deleted file mode 100644
index b71d037..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/BookKeeperTestBase.java
+++ /dev/null
@@ -1,267 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import java.net.InetAddress;
-import java.nio.ByteBuffer;
-import java.io.File;
-import java.io.IOException;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Random;
-
-import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
-import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
-import org.apache.bookkeeper.test.PortManager;
-
-import org.apache.bookkeeper.bookie.Bookie;
-import org.apache.bookkeeper.bookie.BookieException;
-import org.apache.bookkeeper.conf.ClientConfiguration;
-import org.apache.bookkeeper.conf.ServerConfiguration;
-import org.apache.bookkeeper.client.BookKeeper;
-import org.apache.bookkeeper.proto.BookieServer;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import org.apache.hedwig.util.FileUtils;
-import org.apache.hedwig.zookeeper.ZooKeeperTestBase;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This is a base class for any tests that require a BookKeeper client/server
- * setup.
- *
- */
-public class BookKeeperTestBase extends ZooKeeperTestBase {
-    private static final Logger LOG = LoggerFactory.getLogger(BookKeeperTestBase.class);
-
-    class TestBookie extends Bookie {
-        final long readDelay;
-
-        public TestBookie(ServerConfiguration conf, long readDelay)
-            throws IOException, KeeperException, InterruptedException, BookieException {
-            super(conf);
-            this.readDelay = readDelay;
-        }
-
-        @Override
-        public ByteBuffer readEntry(long ledgerId, long entryId)
-            throws IOException, NoLedgerException {
-            if (readDelay > 0) {
-                try {
-                    Thread.sleep(readDelay);
-                } catch (InterruptedException ie) {
-                }
-            }
-            return super.readEntry(ledgerId, entryId);
-        }
-    }
-
-    class TestBookieServer extends BookieServer {
-        public TestBookieServer(ServerConfiguration conf)
-            throws IOException,
-                KeeperException, InterruptedException, BookieException,
-                UnavailableException, CompatibilityException {
-            super(conf);
-        }
-
-        protected Bookie newBookie(ServerConfiguration conf)
-            throws IOException, KeeperException, InterruptedException, BookieException {
-            return new TestBookie(conf, readDelay);
-        }
-    }
-
-    // BookKeeper Server variables
-    private List<BookieServer> bookiesList;
-    private List<ServerConfiguration> bkConfsList;
-
-    // String constants used for creating the bookie server files.
-    private static final String PREFIX = "bookie";
-    private static final String SUFFIX = "test";
-
-    // readDelay
-    protected long readDelay;
-
-    // Variable to decide how many bookie servers to set up.
-    private final int numBookies;
-    // BookKeeper client instance
-    protected BookKeeper bk;
-
-    protected ServerConfiguration baseConf = newServerConfiguration();
-    protected ClientConfiguration baseClientConf = new ClientConfiguration();
-
-    // Constructor
-    public BookKeeperTestBase(int numBookies) {
-        this(numBookies, 0L);
-    }
-
-    public BookKeeperTestBase(int numBookies, long readDelay) {
-        this.numBookies = numBookies;
-        this.readDelay = readDelay;
-    }
-
-    public BookKeeperTestBase() {
-        // By default, use 3 bookies.
-        this(3);
-    }
-
-    // Getter for the ZooKeeper client instance that the parent class sets up.
-    protected ZooKeeper getZooKeeperClient() {
-        return zk;
-    }
-
-    // Give junit a fake test so that its happy
-    @Test(timeout=60000)
-    public void testNothing() throws Exception {
-
-    }
-
-    @Override
-    @Before
-    public void setUp() throws Exception {
-        super.setUp();
-        // Initialize the zk client with values
-        try {
-            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);
-        } catch (KeeperException e) {
-            LOG.error("Error setting up", e);
-        } catch (InterruptedException e) {
-            LOG.error("Error setting up", e);
-        }
-
-        // Create Bookie Servers
-        bookiesList = new LinkedList<BookieServer>();
-        bkConfsList = new LinkedList<ServerConfiguration>();
-
-        for (int i = 0; i < numBookies; i++) {
-            startUpNewBookieServer();
-        }
-
-        // Create the BookKeeper client
-        bk = new BookKeeper(hostPort);
-    }
-
-    public String getZkHostPort() {
-        return hostPort;
-    }
-
-    @Override
-    @After
-    public void tearDown() throws Exception {
-        // Shutdown all of the bookie servers
-        for (BookieServer bs : bookiesList) {
-            bs.shutdown();
-        }
-        // Close the BookKeeper client
-        bk.close();
-        super.tearDown();
-    }
-
-    public void stopAllBookieServers() throws Exception {
-        for (BookieServer bs : bookiesList) {
-            bs.shutdown();
-        }
-        bookiesList.clear();
-    }
-
-    public void startAllBookieServers() throws Exception {
-        for (ServerConfiguration conf : bkConfsList) {
-            bookiesList.add(startBookie(conf));
-        }
-    }
-
-    public void suspendAllBookieServers() throws Exception {
-        for (BookieServer bs : bookiesList) {
-            bs.suspendProcessing();
-        }
-    }
-
-    public void resumeAllBookieServers() throws Exception {
-        for (BookieServer bs : bookiesList) {
-            bs.resumeProcessing();
-        }
-    }
-
-    public void tearDownOneBookieServer() throws Exception {
-        Random r = new Random();
-        int bi = r.nextInt(bookiesList.size());
-        BookieServer bs = bookiesList.get(bi);
-        bs.shutdown();
-        bookiesList.remove(bi);
-        bkConfsList.remove(bi);
-    }
-    
-    public void startUpNewBookieServer() throws Exception {
-        int port = PortManager.nextFreePort();
-        File tmpDir = FileUtils.createTempDirectory(
-                PREFIX + port, SUFFIX);
-        ServerConfiguration conf = newServerConfiguration(
-                port, hostPort, tmpDir, new File[] { tmpDir });
-        bookiesList.add(startBookie(conf));
-        bkConfsList.add(conf);
-    }
-
-    /**
-     * Helper method to startup a bookie server using a configuration object
-     *
-     * @param conf
-     *            Server Configuration Object
-     *
-     */
-    private BookieServer startBookie(ServerConfiguration conf) throws Exception {
-        BookieServer server = new TestBookieServer(conf);
-        server.start();
-
-        int port = conf.getBookiePort();
-        while(zk.exists("/ledgers/available/" + InetAddress.getLocalHost().getHostAddress() + ":" + port, false) == null) {
-            Thread.sleep(500);
-        }
-
-        return server;
-    }
-
-    // construct the basic server configuration for bookkeeper testing
-    private static ServerConfiguration newServerConfiguration() {
-        ServerConfiguration conf = new ServerConfiguration();
-        conf.setJournalFlushWhenQueueEmpty(true);
-        conf.setJournalAdaptiveGroupWrites(false);
-        return conf;
-    }
-
-    protected ServerConfiguration newServerConfiguration(int port, String zkServers, File journalDir, File[] ledgerDirs) {
-        ServerConfiguration conf = new ServerConfiguration(baseConf);
-        conf.setAllowLoopback(true);
-        conf.setBookiePort(port);
-        conf.setZkServers(zkServers);
-        conf.setJournalDirName(journalDir.getPath());
-        String[] ledgerDirNames = new String[ledgerDirs.length];
-        for (int i=0; i<ledgerDirs.length; i++) {
-            ledgerDirNames[i] = ledgerDirs[i].getPath();
-        }
-        conf.setLedgerDirNames(ledgerDirNames);
-        return conf;
-    }
-
-}


Mime
View raw message