bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [06/49] bookkeeper git commit: BOOKKEEPER-769: Remove the Hedwig Code
Date Wed, 16 Mar 2016 03:44:16 GMT
http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java b/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java
deleted file mode 100644
index 0150a11..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java
+++ /dev/null
@@ -1,1320 +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;
-
-import java.net.InetAddress;
-import java.io.File;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import com.google.protobuf.ByteString;
-
-import org.junit.Test;
-import static org.junit.Assert.*;
-
-import org.apache.bookkeeper.test.ZooKeeperUtil;
-import org.apache.bookkeeper.test.PortManager;
-
-import org.apache.hedwig.util.HedwigSocketAddress;
-
-import org.apache.commons.configuration.Configuration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Test Backward Compatability between different versions
- */
-public class TestBackwardCompat {
-
-    private static final Logger logger = LoggerFactory.getLogger(TestBackwardCompat.class);
-
-    static final int CONSUMEINTERVAL = 5;
-    static ZooKeeperUtil zkUtil = new ZooKeeperUtil();
-
-    static class BookKeeperCluster400 {
-
-        int numBookies;
-        List<org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration> bkConfs;
-        List<org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer> bks;
-
-        BookKeeperCluster400(int numBookies) {
-            this.numBookies = numBookies;
-        }
-
-        public void start() throws Exception {
-            zkUtil.startServer();
-
-            bks = new LinkedList<org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer>();
-            bkConfs = new LinkedList<org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration>();
-
-            for (int i=0; i<numBookies; i++) {
-                startBookieServer();
-            }
-        }
-
-        public void stop() throws Exception {
-            for (org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer bs : bks) {
-                bs.shutdown();
-            }
-            bks.clear();
-
-            zkUtil.killServer();
-        }
-
-        protected void startBookieServer() throws Exception {
-            int port = PortManager.nextFreePort();
-            File tmpDir = org.apache.hw_v4_0_0.hedwig.util.FileUtils.createTempDirectory(
-                getClass().getName() + port, "test");
-            org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
-                port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
-            bks.add(startBookie(conf));
-            bkConfs.add(conf);
-        }
-
-        protected org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration newServerConfiguration(
-            int port, String zkServers, File journalDir, File[] ledgerDirs) {
-            org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration conf =
-                new org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration();
-            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;
-        }
-
-        protected org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer startBookie(
-            org.apache.hw_v4_0_0.bookkeeper.conf.ServerConfiguration conf) throws Exception {
-            org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer server
-                = new org.apache.hw_v4_0_0.bookkeeper.proto.BookieServer(conf);
-            server.start();
-
-            int port = conf.getBookiePort();
-            while (zkUtil.getZooKeeperClient().exists(
-                    "/ledgers/available/" + InetAddress.getLocalHost().getHostAddress() + ":" + port,
-                    false) == null) {
-                Thread.sleep(500);
-            }
-            return server;
-        }
-    }
-
-    /**
-     * Version 4.0.0 classes
-     */
-    static class Server400 {
-        org.apache.hw_v4_0_0.hedwig.server.common.ServerConfiguration conf;
-        org.apache.hw_v4_0_0.hedwig.server.netty.PubSubServer server;
-
-        Server400(final String zkHosts, final int port, final int sslPort) {
-            conf = new org.apache.hw_v4_0_0.hedwig.server.common.ServerConfiguration() {
-                @Override
-                public String getZkHost() {
-                    return zkHosts;
-                }
-
-                @Override
-                public int getServerPort() {
-                    return port;
-                }
-
-                @Override
-                public int getSSLServerPort() {
-                    return sslPort;
-                }
-            };
-        }
-
-        void start() throws Exception {
-            server = new org.apache.hw_v4_0_0.hedwig.server.netty.PubSubServer(conf);
-        }
-
-        void stop() throws Exception {
-            if (null != server) {
-                server.shutdown();
-            }
-        }
-    }
-
-    static class Client400 {
-        org.apache.hw_v4_0_0.hedwig.client.conf.ClientConfiguration conf;
-        org.apache.hw_v4_0_0.hedwig.client.api.Client client;
-        org.apache.hw_v4_0_0.hedwig.client.api.Publisher publisher;
-        org.apache.hw_v4_0_0.hedwig.client.api.Subscriber subscriber;
-
-        Client400(final String connectString) {
-            conf = new org.apache.hw_v4_0_0.hedwig.client.conf.ClientConfiguration() {
-                    @Override
-                    protected org.apache.hw_v4_0_0.hedwig.util.HedwigSocketAddress
-                        getDefaultServerHedwigSocketAddress() {
-                        return new org.apache.hw_v4_0_0.hedwig.util.HedwigSocketAddress(connectString);
-                    }
-                };
-            client = new org.apache.hw_v4_0_0.hedwig.client.HedwigClient(conf);
-            publisher = client.getPublisher();
-            subscriber = client.getSubscriber();
-        }
-
-        void close() throws Exception {
-            if (null != client) {
-                client.close();
-            }
-        }
-
-        org.apache.hw_v4_0_0.hedwig.protocol.PubSubProtocol.MessageSeqId publish(
-            ByteString topic, ByteString data) throws Exception {
-            org.apache.hw_v4_0_0.hedwig.protocol.PubSubProtocol.Message message =
-                org.apache.hw_v4_0_0.hedwig.protocol.PubSubProtocol.Message.newBuilder()
-                    .setBody(data).build();
-            publisher.publish(topic, message);
-            return null;
-        }
-    }
-
-    static class BookKeeperCluster410 {
-
-        int numBookies;
-        List<org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration> bkConfs;
-        List<org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer> bks;
-
-        BookKeeperCluster410(int numBookies) {
-            this.numBookies = numBookies;
-        }
-
-        public void start() throws Exception {
-            zkUtil.startServer();
-
-            bks = new LinkedList<org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer>();
-            bkConfs = new LinkedList<org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration>();
-
-            for (int i=0; i<numBookies; i++) {
-                startBookieServer();
-            }
-        }
-
-        public void stop() throws Exception {
-            for (org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer bs : bks) {
-                bs.shutdown();
-            }
-            bks.clear();
-
-            zkUtil.killServer();
-        }
-
-        protected void startBookieServer() throws Exception {
-            int port = PortManager.nextFreePort();
-            File tmpDir = org.apache.hw_v4_1_0.hedwig.util.FileUtils.createTempDirectory(
-                getClass().getName() + port, "test");
-            org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
-                    port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
-            bks.add(startBookie(conf));
-            bkConfs.add(conf);
-        }
-
-        protected org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration newServerConfiguration(
-            int port, String zkServers, File journalDir, File[] ledgerDirs) {
-            org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration conf =
-                new org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration();
-            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;
-        }
-
-        protected org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer startBookie(
-            org.apache.hw_v4_1_0.bookkeeper.conf.ServerConfiguration conf) throws Exception {
-            org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer server
-                = new org.apache.hw_v4_1_0.bookkeeper.proto.BookieServer(conf);
-            server.start();
-
-            int port = conf.getBookiePort();
-            while (zkUtil.getZooKeeperClient().exists(
-                    "/ledgers/available/" + InetAddress.getLocalHost().getHostAddress() + ":" + port,
-                    false) == null) {
-                Thread.sleep(500);
-            }
-            return server;
-        }
-    }
-
-    /**
-     * Version 4.1.0 classes
-     */
-    static class Server410 {
-        org.apache.hw_v4_1_0.hedwig.server.common.ServerConfiguration conf;
-        org.apache.hw_v4_1_0.hedwig.server.netty.PubSubServer server;
-
-        Server410(final String zkHosts, final int port, final int sslPort) {
-            conf = new org.apache.hw_v4_1_0.hedwig.server.common.ServerConfiguration() {
-                @Override
-                public int getConsumeInterval() {
-                    return CONSUMEINTERVAL;
-                }
-                @Override
-                public String getZkHost() {
-                    return zkHosts;
-                }
-
-                @Override
-                public int getServerPort() {
-                    return port;
-                }
-
-                @Override
-                public int getSSLServerPort() {
-                    return sslPort;
-                }
-            };
-        }
-
-        void start() throws Exception {
-            server = new org.apache.hw_v4_1_0.hedwig.server.netty.PubSubServer(conf);
-            server.start();
-        }
-
-        void stop() throws Exception {
-            if (null != server) {
-                server.shutdown();
-            }
-        }
-    }
-
-    static class Client410 {
-        org.apache.hw_v4_1_0.hedwig.client.conf.ClientConfiguration conf;
-        org.apache.hw_v4_1_0.hedwig.client.api.Client client;
-        org.apache.hw_v4_1_0.hedwig.client.api.Publisher publisher;
-        org.apache.hw_v4_1_0.hedwig.client.api.Subscriber subscriber;
-
-        class IntMessageHandler implements org.apache.hw_v4_1_0.hedwig.client.api.MessageHandler {
-            ByteString topic;
-            ByteString subId;
-            int next;
-
-            CountDownLatch latch;
-
-            IntMessageHandler(ByteString t, ByteString s, int start, int num) {
-                this.topic = t;
-                this.subId = s;
-                this.next = start;
-                this.latch = new CountDownLatch(num);
-            }
-
-            @Override
-            public void deliver(ByteString t, ByteString s,
-                                org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message msg,
-                                org.apache.hw_v4_1_0.hedwig.util.Callback<Void> callback, Object context) {
-                if (!t.equals(topic) || !s.equals(subId)) {
-                    return;
-                }
-                int num = Integer.parseInt(msg.getBody().toStringUtf8());
-                if (num == next) {
-                    latch.countDown();
-                    ++next;
-                }
-                callback.operationFinished(context, null);
-            }
-
-            public boolean await(long timeout, TimeUnit unit)
-            throws InterruptedException {
-                return latch.await(timeout, unit);
-            }
-        }
-
-        Client410(final String connectString) {
-            conf = new org.apache.hw_v4_1_0.hedwig.client.conf.ClientConfiguration() {
-                @Override
-                public boolean isAutoSendConsumeMessageEnabled() {
-                    return true;
-                }
-                @Override
-                public int getConsumedMessagesBufferSize() {
-                    return 1;
-                }
-                @Override
-                protected org.apache.hw_v4_1_0.hedwig.util.HedwigSocketAddress
-                    getDefaultServerHedwigSocketAddress() {
-                    return new org.apache.hw_v4_1_0.hedwig.util.HedwigSocketAddress(connectString);
-                }
-            };
-            client = new org.apache.hw_v4_1_0.hedwig.client.HedwigClient(conf);
-            publisher = client.getPublisher();
-            subscriber = client.getSubscriber();
-        }
-
-        void close() throws Exception {
-            if (null != client) {
-                client.close();
-            }
-        }
-
-        org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.MessageSeqId publish(
-            ByteString topic, ByteString data) throws Exception {
-            org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message message =
-                org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message.newBuilder()
-                    .setBody(data).build();
-            publisher.publish(topic, message);
-            return null;
-        }
-
-        void publishInts(ByteString topic, int start, int num) throws Exception {
-            for (int i=0; i<num; i++) {
-                org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message msg =
-                    org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message.newBuilder().setBody(ByteString.copyFromUtf8("" + (start+i))).build();
-                publisher.publish(topic, msg);
-            }
-        }
-
-        void sendXExpectLastY(ByteString topic, ByteString subid, final int x, final int y)
-        throws Exception {
-            for (int i=0; i<x; i++) {
-                publisher.publish(topic, org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message.newBuilder().setBody(
-                                         ByteString.copyFromUtf8(String.valueOf(i))).build());
-            }
-            subscriber.subscribe(topic, subid, org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.ATTACH);
-
-            final AtomicInteger expected = new AtomicInteger(x - y);
-            final CountDownLatch latch = new CountDownLatch(1);
-            subscriber.startDelivery(topic, subid, new org.apache.hw_v4_1_0.hedwig.client.api.MessageHandler() {
-                @Override
-                synchronized public void deliver(ByteString topic, ByteString subscriberId,
-                                                 org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message msg,
-                                                 org.apache.hw_v4_1_0.hedwig.util.Callback<Void> callback, Object context) {
-                    try {
-                        int value = Integer.valueOf(msg.getBody().toStringUtf8());
-                        if (value == expected.get()) {
-                            expected.incrementAndGet();
-                        } else {
-                            logger.error("Did not receive expected value, expected {}, got {}",
-                                         expected.get(), value);
-                            expected.set(0);
-                            latch.countDown();
-                        }
-                        if (expected.get() == x) {
-                            latch.countDown();
-                        }
-                        callback.operationFinished(context, null);
-                    } catch (Exception e) {
-                        logger.error("Received bad message", e);
-                        latch.countDown();
-                    }
-                }
-            });
-            assertTrue("Timed out waiting for messages Y is " + y + " expected is currently "
-                       + expected.get(), latch.await(10, TimeUnit.SECONDS));
-            assertEquals("Should be expected message with " + x, x, expected.get());
-            subscriber.stopDelivery(topic, subid);
-            subscriber.closeSubscription(topic, subid);
-            Thread.sleep(1000); // give server time to run disconnect logic (BOOKKEEPER-513)
-        }
-
-        void subscribe(ByteString topic, ByteString subscriberId) throws Exception {
-            org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options =
-                org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH).build();
-            subscribe(topic, subscriberId, options);
-        }
-
-        void subscribe(ByteString topic, ByteString subscriberId,
-                       org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options) throws Exception {
-            subscriber.subscribe(topic, subscriberId, options);
-        }
-
-        void closeSubscription(ByteString topic, ByteString subscriberId) throws Exception {
-            subscriber.closeSubscription(topic, subscriberId);
-            Thread.sleep(1000); // give server time to run disconnect logic (BOOKKEEPER-513)
-        }
-
-        void receiveInts(ByteString topic, ByteString subscriberId, int start, int num) throws Exception {
-            IntMessageHandler msgHandler = new IntMessageHandler(topic, subscriberId, start, num);
-            subscriber.startDelivery(topic, subscriberId, msgHandler);
-            msgHandler.await(num, TimeUnit.SECONDS);
-            subscriber.stopDelivery(topic, subscriberId);
-        }
-    }
-
-    /**
-     * 4.2.0 Version
-     */
-    static class BookKeeperCluster420{
-
-        int numBookies;
-        List<org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration> bkConfs;
-        List<org.apache.hw_v4_2_0.bookkeeper.proto.BookieServer> bks;
-
-
-        BookKeeperCluster420(int numBookies) {
-            this.numBookies = numBookies;
-        }
-
-        public void start() throws Exception {
-            zkUtil.startServer();
-
-            bks = new LinkedList<org.apache.hw_v4_2_0.bookkeeper.proto.BookieServer>();
-            bkConfs = new LinkedList<org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration>();
-
-            for (int i=0; i<numBookies; i++) {
-                startBookieServer();
-            }
-        }
-
-        public void stop() throws Exception {
-            for (org.apache.hw_v4_2_0.bookkeeper.proto.BookieServer bs : bks) {
-                bs.shutdown();
-            }
-            bks.clear();
-
-            zkUtil.killServer();
-        }
-
-        protected void startBookieServer() throws Exception {
-            int port = PortManager.nextFreePort();
-            File tmpDir = org.apache.hw_v4_2_0.hedwig.util.FileUtils.createTempDirectory(
-                getClass().getName() + port, "test");
-            org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
-                port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
-            bks.add(startBookie(conf));
-            bkConfs.add(conf);
-        }
-
-        protected org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration newServerConfiguration(
-            int port, String zkServers, File journalDir, File[] ledgerDirs) {
-            org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration conf =
-                new org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration();
-            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;
-        }
-
-        protected org.apache.hw_v4_2_0.bookkeeper.proto.BookieServer startBookie(
-            org.apache.hw_v4_2_0.bookkeeper.conf.ServerConfiguration conf) throws Exception {
-            org.apache.hw_v4_2_0.bookkeeper.proto.BookieServer server
-                = new org.apache.hw_v4_2_0.bookkeeper.proto.BookieServer(conf);
-            server.start();
-
-            int port = conf.getBookiePort();
-            while (zkUtil.getZooKeeperClient().exists(
-                    "/ledgers/available/" + InetAddress.getLocalHost().getHostAddress() + ":" + port,
-                    false) == null) {
-                Thread.sleep(500);
-            }
-            return server;
-        }
-    }
-
-    static class Server420 {
-        org.apache.hw_v4_2_0.hedwig.server.common.ServerConfiguration conf;
-        org.apache.hw_v4_2_0.hedwig.server.netty.PubSubServer server;
-
-        Server420(final String zkHosts, final int port, final int sslPort) {
-            conf = new org.apache.hw_v4_2_0.hedwig.server.common.ServerConfiguration() {
-                @Override
-                public int getConsumeInterval() {
-                    return CONSUMEINTERVAL;
-                }
-
-                @Override
-                public String getZkHost() {
-                    return zkHosts;
-                }
-
-                @Override
-                public int getServerPort() {
-                    return port;
-                }
-
-                @Override
-                public int getSSLServerPort() {
-                    return sslPort;
-                }
-            };
-        }
-
-        void start() throws Exception {
-            server = new org.apache.hw_v4_2_0.hedwig.server.netty.PubSubServer(conf);
-            server.start();
-        }
-
-        void stop() throws Exception {
-            if (null != server) {
-                server.shutdown();
-            }
-        }
-    }
-
-    /**
-     * Current Version
-     */
-    static class BookKeeperClusterCurrent {
-
-        int numBookies;
-        List<org.apache.bookkeeper.conf.ServerConfiguration> bkConfs;
-        List<org.apache.bookkeeper.proto.BookieServer> bks;
-
-
-        BookKeeperClusterCurrent(int numBookies) {
-            this.numBookies = numBookies;
-        }
-
-        public void start() throws Exception {
-            zkUtil.startServer();
-
-            bks = new LinkedList<org.apache.bookkeeper.proto.BookieServer>();
-            bkConfs = new LinkedList<org.apache.bookkeeper.conf.ServerConfiguration>();
-
-            for (int i=0; i<numBookies; i++) {
-                startBookieServer();
-            }
-        }
-
-        public void stop() throws Exception {
-            for (org.apache.bookkeeper.proto.BookieServer bs : bks) {
-                bs.shutdown();
-            }
-            bks.clear();
-
-            zkUtil.killServer();
-        }
-
-        protected void startBookieServer() throws Exception {
-            int port = PortManager.nextFreePort();
-            File tmpDir = org.apache.hedwig.util.FileUtils.createTempDirectory(
-                getClass().getName() + port, "test");
-            org.apache.bookkeeper.conf.ServerConfiguration conf = newServerConfiguration(
-                port, zkUtil.getZooKeeperConnectString(), tmpDir, new File[] { tmpDir });
-            conf.setAllowLoopback(true);
-            bks.add(startBookie(conf));
-            bkConfs.add(conf);
-        }
-
-        protected org.apache.bookkeeper.conf.ServerConfiguration newServerConfiguration(
-            int port, String zkServers, File journalDir, File[] ledgerDirs) {
-            org.apache.bookkeeper.conf.ServerConfiguration conf =
-                new org.apache.bookkeeper.conf.ServerConfiguration();
-            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;
-        }
-
-        protected org.apache.bookkeeper.proto.BookieServer startBookie(
-            org.apache.bookkeeper.conf.ServerConfiguration conf) throws Exception {
-            org.apache.bookkeeper.proto.BookieServer server
-                = new org.apache.bookkeeper.proto.BookieServer(conf);
-            server.start();
-
-            int port = conf.getBookiePort();
-            while (zkUtil.getZooKeeperClient().exists(
-                    "/ledgers/available/" + InetAddress.getLocalHost().getHostAddress() + ":" + port,
-                    false) == null) {
-                Thread.sleep(500);
-            }
-            return server;
-        }
-    }
-
-    static class ServerCurrent {
-        org.apache.hedwig.server.common.ServerConfiguration conf;
-        org.apache.hedwig.server.netty.PubSubServer server;
-
-        ServerCurrent(final String zkHosts, final int port, final int sslPort) {
-            conf = new org.apache.hedwig.server.common.ServerConfiguration() {
-                @Override
-                public int getConsumeInterval() {
-                    return CONSUMEINTERVAL;
-                }
-
-                @Override
-                public String getZkHost() {
-                    return zkHosts;
-                }
-
-                @Override
-                public int getServerPort() {
-                    return port;
-                }
-
-                @Override
-                public int getSSLServerPort() {
-                    return sslPort;
-                }
-            };
-        }
-
-        void start() throws Exception {
-            server = new org.apache.hedwig.server.netty.PubSubServer(conf);
-            server.start();
-        }
-
-        void stop() throws Exception {
-            if (null != server) {
-                server.shutdown();
-            }
-        }
-    }
-
-    static class ClientCurrent {
-        org.apache.hedwig.client.conf.ClientConfiguration conf;
-        org.apache.hedwig.client.api.Client client;
-        org.apache.hedwig.client.api.Publisher publisher;
-        org.apache.hedwig.client.api.Subscriber subscriber;
-
-        class IntMessageHandler implements org.apache.hedwig.client.api.MessageHandler {
-            ByteString topic;
-            ByteString subId;
-            int next;
-
-            CountDownLatch latch;
-
-            IntMessageHandler(ByteString t, ByteString s, int start, int num) {
-                this.topic = t;
-                this.subId = s;
-                this.next = start;
-                this.latch = new CountDownLatch(num);
-            }
-
-            @Override
-            public void deliver(ByteString t, ByteString s,
-                                org.apache.hedwig.protocol.PubSubProtocol.Message msg,
-                                org.apache.hedwig.util.Callback<Void> callback, Object context) {
-                if (!t.equals(topic) || !s.equals(subId)) {
-                    return;
-                }
-                int num = Integer.parseInt(msg.getBody().toStringUtf8());
-                if (num == next) {
-                    latch.countDown();
-                    ++next;
-                }
-                callback.operationFinished(context, null);
-            }
-
-            public boolean await(long timeout, TimeUnit unit)
-            throws InterruptedException {
-                return latch.await(timeout, unit);
-            }
-        }
-
-        ClientCurrent(final String connectString) {
-            this(true, connectString);
-        }
-
-        ClientCurrent(final boolean autoConsumeEnabled, final String connectString) {
-            conf = new org.apache.hedwig.client.conf.ClientConfiguration() {
-                @Override
-                public boolean isAutoSendConsumeMessageEnabled() {
-                    return autoConsumeEnabled;
-                }
-                @Override
-                public int getConsumedMessagesBufferSize() {
-                    return 1;
-                }
-                @Override
-                protected HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
-                    return new HedwigSocketAddress(connectString);
-                }
-            };
-            client = new org.apache.hedwig.client.HedwigClient(conf);
-            publisher = client.getPublisher();
-            subscriber = client.getSubscriber();
-        }
-
-        void close() throws Exception {
-            if (null != client) {
-                client.close();
-            }
-        }
-
-        org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId publish(
-            ByteString topic, ByteString data) throws Exception {
-            org.apache.hedwig.protocol.PubSubProtocol.Message message =
-                org.apache.hedwig.protocol.PubSubProtocol.Message.newBuilder()
-                    .setBody(data).build();
-            org.apache.hedwig.protocol.PubSubProtocol.PublishResponse resp =
-                publisher.publish(topic, message);
-            if (null == resp) {
-                return null;
-            }
-            return resp.getPublishedMsgId();
-        }
-
-        void publishInts(ByteString topic, int start, int num) throws Exception {
-            for (int i=0; i<num; i++) {
-                org.apache.hedwig.protocol.PubSubProtocol.Message msg =
-                    org.apache.hedwig.protocol.PubSubProtocol.Message.newBuilder().setBody(ByteString.copyFromUtf8("" + (start+i))).build();
-                publisher.publish(topic, msg);
-            }
-        }
-
-        void sendXExpectLastY(ByteString topic, ByteString subid, final int x, final int y)
-        throws Exception {
-            for (int i=0; i<x; i++) {
-                publisher.publish(topic, org.apache.hedwig.protocol.PubSubProtocol.Message.newBuilder().setBody(
-                                         ByteString.copyFromUtf8(String.valueOf(i))).build());
-            }
-            org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions opts
-                = org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.ATTACH)
-                .build();
-            subscriber.subscribe(topic, subid, opts);
-
-            final AtomicInteger expected = new AtomicInteger(x - y);
-            final CountDownLatch latch = new CountDownLatch(1);
-            subscriber.startDelivery(topic, subid, new org.apache.hedwig.client.api.MessageHandler() {
-                @Override
-                synchronized public void deliver(ByteString topic, ByteString subscriberId,
-                                                 org.apache.hedwig.protocol.PubSubProtocol.Message msg,
-                                                 org.apache.hedwig.util.Callback<Void> callback, Object context) {
-                    try {
-                        int value = Integer.valueOf(msg.getBody().toStringUtf8());
-                        if (value == expected.get()) {
-                            expected.incrementAndGet();
-                        } else {
-                            logger.error("Did not receive expected value, expected {}, got {}",
-                                         expected.get(), value);
-                            expected.set(0);
-                            latch.countDown();
-                        }
-                        if (expected.get() == x) {
-                            latch.countDown();
-                        }
-                        callback.operationFinished(context, null);
-                    } catch (Exception e) {
-                        logger.error("Received bad message", e);
-                        latch.countDown();
-                    }
-                }
-            });
-            assertTrue("Timed out waiting for messages Y is " + y + " expected is currently "
-                       + expected.get(), latch.await(10, TimeUnit.SECONDS));
-            assertEquals("Should be expected message with " + x, x, expected.get());
-            subscriber.stopDelivery(topic, subid);
-            subscriber.closeSubscription(topic, subid);
-        }
-
-        void receiveNumModM(final ByteString topic, final ByteString subid,
-                            final int start, final int num, final int M) throws Exception {
-            org.apache.hedwig.filter.ServerMessageFilter filter =
-                new org.apache.hedwig.filter.ServerMessageFilter() {
-
-                @Override
-                public org.apache.hedwig.filter.ServerMessageFilter
-                    initialize(Configuration conf) {
-                    // do nothing
-                    return this;
-                }
-
-                @Override
-                public void uninitialize() {
-                    // do nothing;
-                }
-
-                @Override
-                public org.apache.hedwig.filter.MessageFilterBase
-                    setSubscriptionPreferences(ByteString topic, ByteString subscriberId,
-                    org.apache.hedwig.protocol.PubSubProtocol.SubscriptionPreferences preferences) {
-                    // do nothing;
-                    return this;
-                }
-
-                @Override
-                public boolean testMessage(org.apache.hedwig.protocol.PubSubProtocol.Message msg) {
-                    int value = Integer.valueOf(msg.getBody().toStringUtf8());
-                    return 0 == value % M;
-                }
-            };
-            filter.initialize(conf.getConf());
-
-
-            org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions opts
-                = org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.ATTACH)
-                .build();
-            subscriber.subscribe(topic, subid, opts);
-            final int base = start + M - start % M;
-            final AtomicInteger expected = new AtomicInteger(base);
-            final CountDownLatch latch = new CountDownLatch(1);
-            subscriber.startDeliveryWithFilter(topic, subid, new org.apache.hedwig.client.api.MessageHandler() {
-                synchronized public void deliver(ByteString topic, ByteString subscriberId,
-                                                 org.apache.hedwig.protocol.PubSubProtocol.Message msg,
-                                                 org.apache.hedwig.util.Callback<Void> callback, Object context) {
-                    try {
-                        int value = Integer.valueOf(msg.getBody().toStringUtf8());
-                        // duplicated messages received, ignore them
-                        if (value > start) {
-                            if (value == expected.get()) {
-                                expected.addAndGet(M);
-                            } else {
-                                logger.error("Did not receive expected value, expected {}, got {}",
-                                             expected.get(), value);
-                                expected.set(0);
-                                latch.countDown();
-                            }
-                            if (expected.get() == (base + num * M)) {
-                                latch.countDown();
-                            }
-                        }
-                        callback.operationFinished(context, null);
-                    } catch (Exception e) {
-                        logger.error("Received bad message", e);
-                        latch.countDown();
-                    }
-                }
-            }, (org.apache.hedwig.filter.ClientMessageFilter) filter);
-            assertTrue("Timed out waiting for messages mod " + M + " expected is " + expected.get(),
-                       latch.await(10, TimeUnit.SECONDS));
-            assertEquals("Should be expected message with " + (base + num * M), (base + num*M), expected.get());
-            subscriber.stopDelivery(topic, subid);
-            filter.uninitialize();
-            subscriber.closeSubscription(topic, subid);
-        }
-
-        void subscribe(ByteString topic, ByteString subscriberId) throws Exception {
-            org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options =
-                org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH).build();
-            subscribe(topic, subscriberId, options);
-        }
-
-        void subscribe(ByteString topic, ByteString subscriberId,
-                       org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options) throws Exception {
-            subscriber.subscribe(topic, subscriberId, options);
-        }
-
-        void closeSubscription(ByteString topic, ByteString subscriberId) throws Exception {
-            subscriber.closeSubscription(topic, subscriberId);
-        }
-
-        void receiveInts(ByteString topic, ByteString subscriberId, int start, int num) throws Exception {
-            IntMessageHandler msgHandler = new IntMessageHandler(topic, subscriberId, start, num);
-            subscriber.startDelivery(topic, subscriberId, msgHandler);
-            msgHandler.await(num, TimeUnit.SECONDS);
-            subscriber.stopDelivery(topic, subscriberId);
-        }
-
-        // throttle doesn't work talking with 41 server
-        void throttleX41(ByteString topic, ByteString subid, final int X)
-        throws Exception {
-            org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options =
-                org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
-                .setMessageWindowSize(X) .build();
-            subscribe(topic, subid, options);
-            closeSubscription(topic, subid);
-            publishInts(topic, 1, 3*X);
-            subscribe(topic, subid);
-
-            final AtomicInteger expected = new AtomicInteger(1);
-            final CountDownLatch throttleLatch = new CountDownLatch(1);
-            final CountDownLatch nonThrottleLatch = new CountDownLatch(1);
-            subscriber.startDelivery(topic, subid, new org.apache.hedwig.client.api.MessageHandler() {
-                @Override
-                public synchronized void deliver(ByteString topic, ByteString subscriberId,
-                                                 org.apache.hedwig.protocol.PubSubProtocol.Message msg,
-                                                 org.apache.hedwig.util.Callback<Void> callback, Object context) {
-                    try {
-                        int value = Integer.valueOf(msg.getBody().toStringUtf8());
-                        logger.debug("Received message {},", value);
-
-                        if (value == expected.get()) {
-                            expected.incrementAndGet();
-                        } else {
-                            // error condition
-                            logger.error("Did not receive expected value, expected {}, got {}",
-                                         expected.get(), value);
-                            expected.set(0);
-                            throttleLatch.countDown();
-                            nonThrottleLatch.countDown();
-                        }
-                        if (expected.get() > X+1) {
-                            throttleLatch.countDown();
-                        }
-                        if (expected.get() == (3 * X + 1)) {
-                            nonThrottleLatch.countDown();
-                        }
-                        callback.operationFinished(context, null);
-                    } catch (Exception e) {
-                        logger.error("Received bad message", e);
-                        throttleLatch.countDown();
-                        nonThrottleLatch.countDown();
-                    }
-                }
-            });
-            assertTrue("Should Receive more messages than throttle value " + X,
-                        throttleLatch.await(10, TimeUnit.SECONDS));
-
-            assertTrue("Timed out waiting for messages " + (3*X + 1),
-                       nonThrottleLatch.await(10, TimeUnit.SECONDS));
-            assertEquals("Should be expected message with " + (3*X + 1),
-                         3*X + 1, expected.get());
-
-            subscriber.stopDelivery(topic, subid);
-            closeSubscription(topic, subid);
-        }
-    }
-
-    /**
-     * Test compatability of message bound between version 4.0.0 and
-     * current version.
-     *
-     * 1) message bound doesn't take effects on 4.0.0 server.
-     * 2) message bound take effects on both 4.1.0 and current server
-     */
-    @Test(timeout=60000)
-    public void testMessageBoundCompat() throws Exception {
-        ByteString topic = ByteString.copyFromUtf8("testMessageBoundCompat");
-        ByteString subid = ByteString.copyFromUtf8("mysub");
-
-        int port = PortManager.nextFreePort();
-        int sslPort = PortManager.nextFreePort();
-
-        // start bookkeeper 400
-        BookKeeperCluster400 bkc400 = new BookKeeperCluster400(3);
-        bkc400.start();
-
-        // start 400 server
-        Server400 s400 = new Server400(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s400.start();
-
-        org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5cur =
-            org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-            .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
-            .setMessageBound(5).build();
-
-        ClientCurrent ccur = new ClientCurrent("localhost:" + port + ":" + sslPort);
-        ccur.subscribe(topic, subid, options5cur);
-        ccur.closeSubscription(topic, subid);
-        ccur.sendXExpectLastY(topic, subid, 50, 50);
-
-        // stop 400 servers
-        s400.stop();
-        bkc400.stop();
-
-        // start bookkeeper 410
-        BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
-        bkc410.start();
-
-        // start 410 server
-        Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s410.start();
-
-        ccur.subscribe(topic, subid, options5cur);
-        ccur.closeSubscription(topic, subid);
-        ccur.sendXExpectLastY(topic, subid, 50, 5);
-
-        // stop 410 servers
-        s410.stop();
-        bkc410.stop();
-
-        // start bookkeeper current
-        BookKeeperCluster420 bkc420 = new BookKeeperCluster420(3);
-        bkc420.start();
-
-        // start 420 server
-        Server420 s420 = new Server420(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s420.start();
-
-        ccur.subscribe(topic, subid, options5cur);
-        ccur.closeSubscription(topic, subid);
-        ccur.sendXExpectLastY(topic, subid, 50, 5);
-
-        // stop 420 server
-        s420.stop();
-        bkc420.stop();
-
-        ccur.close();
-    }
-
-    /**
-     * Test compatability of publish interface between version 4.1.0
-     * and current verison.
-     *
-     * 1) 4.1.0 client could talk with current server.
-     * 2) current client could talk with 4.1.0 server,
-     *    but no message seq id would be returned
-     */
-    @Test(timeout=60000)
-    public void testPublishCompat410() throws Exception {
-        ByteString topic = ByteString.copyFromUtf8("TestPublishCompat410");
-        ByteString data = ByteString.copyFromUtf8("testdata");
-
-        // start bookkeeper 410
-        BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
-        bkc410.start();
-
-        int port = PortManager.nextFreePort();
-        int sslPort = PortManager.nextFreePort();
-
-        // start 410 server
-        Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s410.start();
-
-        ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
-        Client410 c410 = new Client410("localhost:"+port+":"+sslPort);
-
-        // client c410 could publish message to 410 server
-        assertNull(c410.publish(topic, data));
-        // client ccur could publish message to 410 server
-        // but no message seq id would be returned
-        assertNull(ccur.publish(topic, data));
-
-        // stop 410 server
-        s410.stop();
-
-        // start 420 server
-        Server420 s420 = new Server420(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s420.start();
-
-        // client c410 could publish message to 410 server
-        // but no message seq id would be returned
-        assertNull(c410.publish(topic, data));
-        // client ccur could publish message to current server
-        assertNotNull(ccur.publish(topic, data));
-
-        ccur.close();
-        c410.close();
-
-        // stop 420 server
-        s420.stop();
-        bkc410.stop();
-    }
-
-    /**
-     * Test compatability between version 4.1.0 and the current version.
-     *
-     * A current server could read subscription data recorded by 4.1.0 server.
-     */
-    @Test(timeout=60000)
-    public void testSubscriptionDataCompat410() throws Exception {
-        ByteString topic = ByteString.copyFromUtf8("TestCompat410");
-        ByteString sub410 = ByteString.copyFromUtf8("sub410");
-        ByteString subcur = ByteString.copyFromUtf8("subcur");
-
-        // start bookkeeper 410
-        BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
-        bkc410.start();
-
-        int port = PortManager.nextFreePort();
-        int sslPort = PortManager.nextFreePort();
-
-        // start 410 server
-        Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s410.start();
-
-        Client410 c410 = new Client410("localhost:"+port+":"+sslPort);
-        c410.subscribe(topic, sub410);
-        c410.closeSubscription(topic, sub410);
-        Thread.sleep(1000); // give server time to run disconnect logic (BOOKKEEPER-513)
-
-        ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
-        ccur.subscribe(topic, subcur);
-        ccur.closeSubscription(topic, subcur);
-
-        // publish messages using old client
-        c410.publishInts(topic, 0, 10);
-        // stop 410 server
-        s410.stop();
-
-        // start 420 server
-        Server420 s420 = new Server420(zkUtil.getZooKeeperConnectString(),
-                                               port, sslPort);
-        s420.start();
-
-        c410.subscribe(topic, sub410);
-        c410.receiveInts(topic, sub410, 0, 10);
-
-        ccur.subscribe(topic, subcur);
-        ccur.receiveInts(topic, subcur, 0, 10);
-
-        // publish messages using current client
-        ccur.publishInts(topic, 10, 10);
-
-        c410.receiveInts(topic, sub410, 10, 10);
-        ccur.receiveInts(topic, subcur, 10, 10);
-
-        // stop 420 server
-        s420.stop();
-
-        c410.close();
-        ccur.close();
-
-        // stop bookkeeper cluster
-        bkc410.stop();
-    }
-
-    /**
-     * Test compatability between version 4.1.0 and the current version.
-     *
-     * A 4.1.0 client could not update message bound, while current could do it.
-     */
-    @Test(timeout=60000)
-    public void testUpdateMessageBoundCompat410() throws Exception {
-        ByteString topic = ByteString.copyFromUtf8("TestUpdateMessageBoundCompat410");
-        ByteString subid = ByteString.copyFromUtf8("mysub");
-
-        // start bookkeeper
-        BookKeeperCluster420 bkc420 = new BookKeeperCluster420(3);
-        bkc420.start();
-
-        int port = PortManager.nextFreePort();
-        int sslPort = PortManager.nextFreePort();
-
-        // start hub server
-        Server420 s420 = new Server420(zkUtil.getZooKeeperConnectString(),
-                                               port, sslPort);
-        s420.start();
-
-        org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5cur =
-            org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-            .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
-            .setMessageBound(5).build();
-        org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5v410 =
-            org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-            .setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
-            .setMessageBound(5).build();
-        org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options20v410 =
-            org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
-            .setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
-            .setMessageBound(20).build();
-
-        Client410 c410 = new Client410("localhost:"+port+":"+sslPort);
-        c410.subscribe(topic, subid, options20v410);
-        c410.closeSubscription(topic, subid);
-        Thread.sleep(1000); // give server time to run disconnect logic (BOOKKEEPER-513)
-
-        c410.sendXExpectLastY(topic, subid, 50, 20);
-
-        c410.subscribe(topic, subid, options5v410);
-        c410.closeSubscription(topic, subid);
-        Thread.sleep(1000); // give server time to run disconnect logic (BOOKKEEPER-513)
-
-        // the message bound isn't updated.
-        c410.sendXExpectLastY(topic, subid, 50, 20);
-
-        ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
-        ccur.subscribe(topic, subid, options5cur);
-        ccur.closeSubscription(topic, subid);
-        Thread.sleep(1000); // give server time to run disconnect logic (BOOKKEEPER-513)
-
-        // the message bound should be updated.
-        c410.sendXExpectLastY(topic, subid, 50, 5);
-
-        // stop 420 server
-        s420.stop();
-
-        c410.close();
-        ccur.close();
-
-        // stop bookkeeper cluster
-        bkc420.stop();
-    }
-
-    /**
-     * Test compatability between version 4.1.0 and the current version.
-     *
-     * A current client running message filter would fail on 4.1.0 hub servers.
-     */
-    @Test(timeout=60000)
-    public void testClientMessageFilterCompat410() throws Exception {
-        ByteString topic = ByteString.copyFromUtf8("TestUpdateMessageBoundCompat410");
-        ByteString subid = ByteString.copyFromUtf8("mysub");
-
-        // start bookkeeper
-        BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
-        bkc410.start();
-
-        int port = PortManager.nextFreePort();
-        int sslPort = PortManager.nextFreePort();
-
-        // start hub server 410
-        Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s410.start();
-
-        ClientCurrent ccur = new ClientCurrent("localhost:"+port+":"+sslPort);
-        ccur.subscribe(topic, subid);
-        ccur.closeSubscription(topic, subid);
-
-        ccur.publishInts(topic, 0, 100);
-        try {
-            ccur.receiveNumModM(topic, subid, 0, 50, 2);
-            fail("client-side filter could not run on 4.1.0 hub server");
-        } catch (Exception e) {
-            logger.info("Should fail to run client-side message filter on 4.1.0 hub server.", e);
-            ccur.closeSubscription(topic, subid);
-        }
-
-        // stop 410 server
-        s410.stop();
-        // stop bookkeeper cluster
-        bkc410.stop();
-    }
-
-    /**
-     * Test compatability between version 4.1.0 and the current version.
-     *
-     * Server side throttling does't work when current client connects to old version
-     * server.
-     */
-    @Test(timeout=60000)
-    public void testServerSideThrottleCompat410() throws Exception {
-        ByteString topic = ByteString.copyFromUtf8("TestServerSideThrottleCompat410");
-        ByteString subid = ByteString.copyFromUtf8("mysub");
-
-        // start bookkeeper
-        BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
-        bkc410.start();
-
-        int port = PortManager.nextFreePort();
-        int sslPort = PortManager.nextFreePort();
-
-        // start hub server 410
-        Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString(), port, sslPort);
-        s410.start();
-
-        ClientCurrent ccur = new ClientCurrent(false, "localhost:"+port+":"+sslPort);
-        ccur.throttleX41(topic, subid, 10);
-
-        ccur.close();
-
-        // stop 410 server
-        s410.stop();
-        // stop bookkeeper cluster
-        bkc410.stop();
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java b/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java
deleted file mode 100644
index 632ea43..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/TestPubSubServerStartup.java
+++ /dev/null
@@ -1,138 +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;
-
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.MalformedURLException;
-
-import junit.framework.Assert;
-
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.server.LoggingExceptionHandler;
-import org.apache.bookkeeper.test.PortManager;
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.server.netty.PubSubServer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.server.NIOServerCnxnFactory;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.Test;
-import org.apache.hedwig.util.FileUtils;
-
-public class TestPubSubServerStartup {
-
-    private static final Logger logger = LoggerFactory.getLogger(TestPubSubServerStartup.class);
-
-    /**
-     * Start-up zookeeper + pubsubserver reading from a config URL. Then stop
-     * and cleanup.
-     *
-     * Loop over that.
-     *
-     * If the pubsub server does not wait for its zookeeper client to be
-     * connected, the pubsub server will fail at startup.
-     *
-     */
-    @Test(timeout=60000)
-    public void testPubSubServerInstantiationWithConfig() throws Exception {
-        for (int i = 0; i < 10; i++) {
-            logger.info("iteration " + i);
-            instantiateAndDestroyPubSubServer();
-        }
-    }
-
-    private void instantiateAndDestroyPubSubServer() throws IOException, InterruptedException, ConfigurationException,
-        MalformedURLException, Exception {
-        int zkPort = PortManager.nextFreePort();
-        int hwPort = PortManager.nextFreePort();
-        int hwSSLPort = PortManager.nextFreePort();
-        String hedwigParams = "default_server_host=localhost:" + hwPort + "\n"
-            + "zk_host=localhost:" + zkPort + "\n"
-            + "server_port=" + hwPort + "\n"
-            + "ssl_server_port=" + hwSSLPort + "\n"
-            + "zk_timeout=2000\n";
-
-        File hedwigConfigFile = new File(System.getProperty("java.io.tmpdir") + "/hedwig.cfg");
-        writeStringToFile(hedwigParams, hedwigConfigFile);
-
-        ClientBase.setupTestEnv();
-        File zkTmpDir = FileUtils.createTempDirectory("zookeeper", "test");
-
-        ZooKeeperServer zks = new ZooKeeperServer(zkTmpDir, zkTmpDir, zkPort);
-
-        NIOServerCnxnFactory serverFactory = new NIOServerCnxnFactory();
-        serverFactory.configure(new InetSocketAddress(zkPort), 100);
-        serverFactory.startup(zks);
-
-        boolean b = ClientBase.waitForServerUp("127.0.0.1:" + zkPort, 5000);
-        ServerConfiguration serverConf = new ServerConfiguration();
-        serverConf.loadConf(hedwigConfigFile.toURI().toURL());
-
-        logger.info("Zookeeper server up and running!");
-
-        ZooKeeper zkc = new ZooKeeper("127.0.0.1:" + zkPort, 5000, null);
-
-        // initialize the zk client with (fake) values
-        zkc.create("/ledgers", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-        zkc.create("/ledgers/available", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-
-        zkc.close();
-        PubSubServer hedwigServer = null;
-        try {
-            logger.info("starting hedwig broker!");
-            hedwigServer = new PubSubServer(serverConf, new ClientConfiguration(), new LoggingExceptionHandler());
-            hedwigServer.start();
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-        Assert.assertNotNull("failed to instantiate hedwig pub sub server", hedwigServer);
-
-        hedwigServer.shutdown();
-        serverFactory.shutdown();
-
-        zks.shutdown();
-
-        zkTmpDir.delete();
-
-        ClientBase.waitForServerDown("localhost:" + zkPort, 10000);
-
-    }
-
-    public static void writeStringToFile(String string, File f) throws IOException {
-        if (f.exists()) {
-            if (!f.delete()) {
-                throw new RuntimeException("cannot create file " + f.getAbsolutePath());
-            }
-        }
-        if (!f.createNewFile()) {
-            throw new RuntimeException("cannot create new file " + f.getAbsolutePath());
-        }
-
-        FileWriter fw = new FileWriter(f);
-        fw.write(string);
-        fw.close();
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java b/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java
deleted file mode 100644
index 978649a..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java
+++ /dev/null
@@ -1,90 +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.delivery;
-
-import java.util.LinkedList;
-import java.util.Queue;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.client.data.TopicSubscriber;
-import org.apache.hedwig.filter.ServerMessageFilter;
-import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionEvent;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionPreferences;
-import org.apache.hedwig.util.Callback;
-
-public class StubDeliveryManager implements DeliveryManager {
-
-    public static class StartServingRequest {
-        public ByteString topic;
-        public ByteString subscriberId;
-        public MessageSeqId seqIdToStartFrom;
-        public DeliveryEndPoint endPoint;
-        public ServerMessageFilter filter;
-
-        public StartServingRequest(ByteString topic, ByteString subscriberId,
-                                   SubscriptionPreferences preferences,
-                                   MessageSeqId seqIdToStartFrom,
-                                   DeliveryEndPoint endPoint,
-                                   ServerMessageFilter filter) {
-            this.topic = topic;
-            this.subscriberId = subscriberId;
-            this.seqIdToStartFrom = seqIdToStartFrom;
-            this.endPoint = endPoint;
-            this.filter = filter;
-        }
-
-    }
-
-    public Queue<Object> lastRequest = new LinkedList<Object>();
-
-    @Override
-    public void startServingSubscription(ByteString topic, ByteString subscriberId,
-                                         SubscriptionPreferences preferences,
-                                         MessageSeqId seqIdToStartFrom,
-                                         DeliveryEndPoint endPoint,
-                                         ServerMessageFilter filter,
-                                         Callback<Void> cb, Object ctx) {
-        lastRequest.add(new StartServingRequest(topic, subscriberId, preferences,
-                                                seqIdToStartFrom, endPoint, filter));
-        cb.operationFinished(ctx, null);
-    }
-
-    @Override
-    public void stopServingSubscriber(ByteString topic, ByteString subscriberId,
-                                      SubscriptionEvent event,
-                                      Callback<Void> cb, Object ctx) {
-        lastRequest.add(new TopicSubscriber(topic, subscriberId));
-        cb.operationFinished(ctx, null);
-    }
-
-    @Override
-    public void messageConsumed(ByteString topic, ByteString subscriberId,
-                                MessageSeqId seqId) {
-        // do nothing
-    }
-
-    @Override
-    public void start() {
-    }
-
-    @Override
-    public void stop() {
-        // do nothing now
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestFIFODeliveryManager.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestFIFODeliveryManager.java b/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestFIFODeliveryManager.java
deleted file mode 100644
index ebc26f1..0000000
--- a/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/TestFIFODeliveryManager.java
+++ /dev/null
@@ -1,298 +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.delivery;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.filter.PipelineFilter;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionPreferences;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.server.topics.StubTopicManager;
-import org.apache.hedwig.server.persistence.PersistRequest;
-import org.apache.hedwig.server.persistence.PersistenceManager;
-import org.apache.hedwig.server.persistence.StubPersistenceManager;
-import org.apache.hedwig.server.subscriptions.AllToAllTopologyFilter;
-import org.apache.hedwig.util.Callback;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.protobuf.ByteString;
-
-public class TestFIFODeliveryManager {
-    private static final Logger logger = LoggerFactory.getLogger(TestFIFODeliveryManager.class);
-
-    static class TestCallback implements Callback<MessageSeqId> {
-        AtomicBoolean success = new AtomicBoolean(false);
-        final CountDownLatch latch;
-        MessageSeqId msgid = null;
-
-        TestCallback(CountDownLatch l) {
-            this.latch = l;
-        }
-        public void operationFailed(Object ctx, PubSubException exception) {
-            logger.error("Persist operation failed", exception);
-            latch.countDown();
-        }
-
-        public void operationFinished(Object ctx, MessageSeqId resultOfOperation) {
-            msgid = resultOfOperation;
-            success.set(true);
-            latch.countDown();
-        }
-
-        MessageSeqId getId() {
-            assertTrue("Persist operation failed", success.get());
-            return msgid;
-        }
-    }
-
-    /**
-     * Delivery endpoint which puts all responses on a queue
-     */
-    static class ExecutorDeliveryEndPointWithQueue implements DeliveryEndPoint {
-        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-        AtomicInteger numResponses = new AtomicInteger(0);
-        ConcurrentLinkedQueue<PubSubResponse> queue = new ConcurrentLinkedQueue<PubSubResponse>();
-
-        public void send(final PubSubResponse response, final DeliveryCallback callback) {
-            logger.info("Received response {}", response);
-            queue.add(response);
-            numResponses.incrementAndGet();
-            executor.submit(new Runnable() {
-                    public void run() {
-                        callback.sendingFinished();
-                    }
-                });
-        }
-
-        public void close() {
-            executor.shutdown();
-        }
-
-        PubSubResponse getNextResponse() {
-            return queue.poll();
-        }
-
-        int getNumResponses() {
-            return numResponses.get();
-        }
-    }
-
-    /**
-     * Test that the FIFO delivery manager executes stopServing and startServing
-     * in the correct order
-     * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-539}
-     */
-    @Test(timeout = 60000)
-    public void testFIFODeliverySubCloseSubRace() throws Exception {
-        ServerConfiguration conf = new ServerConfiguration();
-        ByteString topic = ByteString.copyFromUtf8("subRaceTopic");
-        ByteString subscriber = ByteString.copyFromUtf8("subRaceSubscriber");
-
-        PersistenceManager pm = new StubPersistenceManager();
-        FIFODeliveryManager fdm = new FIFODeliveryManager(new StubTopicManager(conf), pm, conf);
-        ExecutorDeliveryEndPointWithQueue dep = new ExecutorDeliveryEndPointWithQueue();
-        SubscriptionPreferences prefs = SubscriptionPreferences.newBuilder().build();
-
-        PipelineFilter filter = new PipelineFilter();
-        filter.addLast(new AllToAllTopologyFilter());
-        filter.initialize(conf.getConf());
-        filter.setSubscriptionPreferences(topic, subscriber, prefs);
-        MessageSeqId startId = MessageSeqId.newBuilder().setLocalComponent(1).build();
-
-        CountDownLatch l = new CountDownLatch(1);
-        Message m = Message.newBuilder().setBody(ByteString.copyFromUtf8(String.valueOf(1))).build();
-        TestCallback cb = new TestCallback(l);
-        pm.persistMessage(new PersistRequest(topic, m, cb, null));
-        assertTrue("Persistence never finished", l.await(10, TimeUnit.SECONDS));
-
-        final CountDownLatch oplatch = new CountDownLatch(3);
-        fdm.start();
-        fdm.startServingSubscription(topic, subscriber, prefs, startId, dep, filter,
-                new Callback<Void>() {
-                     @Override
-                     public void operationFinished(Object ctx, Void result) {
-                         oplatch.countDown();
-                     }
-                     @Override
-                     public void operationFailed(Object ctx, PubSubException exception) {
-                         oplatch.countDown();
-                     }
-                }, null);
-        fdm.stopServingSubscriber(topic, subscriber, null,
-                new Callback<Void>() {
-                     @Override
-                     public void operationFinished(Object ctx, Void result) {
-                         oplatch.countDown();
-                     }
-                     @Override
-                     public void operationFailed(Object ctx, PubSubException exception) {
-                         oplatch.countDown();
-                     }
-                }, null);
-        fdm.startServingSubscription(topic, subscriber, prefs, startId, dep, filter,
-                new Callback<Void>() {
-                     @Override
-                     public void operationFinished(Object ctx, Void result) {
-                         oplatch.countDown();
-                     }
-                     @Override
-                     public void operationFailed(Object ctx, PubSubException exception) {
-                         oplatch.countDown();
-                     }
-                }, null);
-
-        assertTrue("Ops never finished", oplatch.await(10, TimeUnit.SECONDS));
-        int seconds = 5;
-        while (dep.getNumResponses() < 2) {
-            if (seconds-- == 0) {
-                break;
-            }
-            Thread.sleep(1000);
-        }
-        PubSubResponse r = dep.getNextResponse();
-        assertNotNull("There should be a response", r);
-        assertTrue("Response should contain a message", r.hasMessage());
-        r = dep.getNextResponse();
-        assertNotNull("There should be a response", r);
-        assertTrue("Response should contain a message", r.hasMessage());
-        r = dep.getNextResponse();
-        assertNull("There should only be 2 responses", r);
-    }
-
-    static class ExecutorDeliveryEndPoint implements DeliveryEndPoint {
-        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-        AtomicInteger numDelivered = new AtomicInteger();
-        final DeliveryManager dm;
-
-        ExecutorDeliveryEndPoint(DeliveryManager dm) {
-            this.dm = dm;
-        }
-
-        public void send(final PubSubResponse response, final DeliveryCallback callback) {
-            executor.submit(new Runnable() {
-                    public void run() {
-                        if (response.hasMessage()) {
-                            MessageSeqId msgid = response.getMessage().getMsgId();
-                            if ((msgid.getLocalComponent() % 2) == 1) {
-                                dm.messageConsumed(response.getTopic(),
-                                        response.getSubscriberId(),
-                                        response.getMessage().getMsgId());
-                            } else {
-                                executor.schedule(new Runnable() {
-                                        public void run() {
-                                            dm.messageConsumed(response.getTopic(),
-                                                    response.getSubscriberId(),
-                                                    response.getMessage().getMsgId());
-                                        }
-                                    }, 1, TimeUnit.SECONDS);
-                            }
-                        }
-                        numDelivered.incrementAndGet();
-                        callback.sendingFinished();
-                    }
-                });
-        }
-
-        public void close() {
-            executor.shutdown();
-        }
-
-        int getNumDelivered() {
-            return numDelivered.get();
-        }
-    }
-
-    /**
-     * Test throttle race issue cause by messageConsumed and doDeliverNextMessage
-     * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-503}
-     */
-    @Test(timeout = 60000)
-    public void testFIFODeliveryThrottlingRace() throws Exception {
-        final int numMessages = 20;
-        final int throttleSize = 10;
-        ServerConfiguration conf = new ServerConfiguration() {
-                @Override
-                public int getDefaultMessageWindowSize() {
-                    return throttleSize;
-                }
-            };
-        ByteString topic = ByteString.copyFromUtf8("throttlingRaceTopic");
-        ByteString subscriber = ByteString.copyFromUtf8("throttlingRaceSubscriber");
-
-        PersistenceManager pm = new StubPersistenceManager();
-        FIFODeliveryManager fdm = new FIFODeliveryManager(new StubTopicManager(conf), pm, conf);
-        ExecutorDeliveryEndPoint dep = new ExecutorDeliveryEndPoint(fdm);
-        SubscriptionPreferences prefs = SubscriptionPreferences.newBuilder().build();
-
-        PipelineFilter filter = new PipelineFilter();
-        filter.addLast(new AllToAllTopologyFilter());
-        filter.initialize(conf.getConf());
-        filter.setSubscriptionPreferences(topic, subscriber, prefs);
-
-        CountDownLatch l = new CountDownLatch(numMessages);
-
-        TestCallback firstCallback = null;
-        for (int i = 0; i < numMessages; i++) {
-            Message m = Message.newBuilder().setBody(ByteString.copyFromUtf8(String.valueOf(i))).build();
-            TestCallback cb = new TestCallback(l);
-            if (firstCallback == null) {
-                firstCallback = cb;
-            }
-            pm.persistMessage(new PersistRequest(topic, m, cb, null));
-        }
-        fdm.start();
-        assertTrue("Persistence never finished", l.await(10, TimeUnit.SECONDS));
-        fdm.startServingSubscription(topic, subscriber, prefs, firstCallback.getId(), dep, filter,
-                new Callback<Void>() {
-                     @Override
-                     public void operationFinished(Object ctx, Void result) {
-                     }
-                     @Override
-                     public void operationFailed(Object ctx, PubSubException exception) {
-                         // would not happened
-                     }
-                }, null);
-
-        int count = 30; // wait for 30 seconds maximum
-        while (dep.getNumDelivered() < numMessages) {
-            Thread.sleep(1000);
-            if (count-- == 0) {
-                break;
-            }
-        }
-        assertEquals("Should have delivered " + numMessages, numMessages, dep.getNumDelivered());
-    }
-
-}


Mime
View raw message