bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [25/49] bookkeeper git commit: BOOKKEEPER-769: Remove the Hedwig Code
Date Wed, 16 Mar 2016 03:44:35 GMT
http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkSubscriber.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkSubscriber.java b/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkSubscriber.java
deleted file mode 100644
index 38dfc3c..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkSubscriber.java
+++ /dev/null
@@ -1,148 +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.client.benchmark;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.Callable;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.protobuf.ByteString;
-
-import org.apache.bookkeeper.util.MathUtils;
-import org.apache.hedwig.client.api.MessageHandler;
-import org.apache.hedwig.client.api.Subscriber;
-import org.apache.hedwig.client.benchmark.BenchmarkUtils.BenchmarkCallback;
-import org.apache.hedwig.client.benchmark.BenchmarkUtils.ThroughputAggregator;
-import org.apache.hedwig.client.benchmark.BenchmarkUtils.ThroughputLatencyAggregator;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.protocol.PubSubProtocol.RegionSpecificSeqId;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions;
-import org.apache.hedwig.util.Callback;
-
-public class BenchmarkSubscriber extends BenchmarkWorker implements Callable<Void> {
-    private static final Logger logger = LoggerFactory.getLogger(BenchmarkSubscriber.class);
-    Subscriber subscriber;
-    ByteString subId;
-
-
-    public BenchmarkSubscriber(int numTopics, int numMessages, int numRegions,
-                               int startTopicLabel, int partitionIndex, int numPartitions, Subscriber subscriber, ByteString subId) {
-        super(numTopics, numMessages, numRegions, startTopicLabel, partitionIndex, numPartitions);
-        this.subscriber = subscriber;
-        this.subId = subId;
-    }
-
-    public void warmup(int numWarmup) throws InterruptedException {
-        /*
-         * multiplying the number of ops by numParitions because we end up
-         * skipping many because of the partitioning logic
-         */
-        multiSub("warmup", "warmup", 0, numWarmup, numWarmup * numPartitions);
-    }
-
-    public Void call() throws Exception {
-
-        final ThroughputAggregator agg = new ThroughputAggregator("recvs", numMessages);
-        agg.startProgress();
-
-        final Map<String, Long> lastSeqIdSeenMap = new HashMap<String, Long>();
-
-        for (int i = startTopicLabel; i < startTopicLabel + numTopics; i++) {
-
-            if (!HedwigBenchmark.amIResponsibleForTopic(i, partitionIndex, numPartitions)) {
-                continue;
-            }
-
-            final String topic = HedwigBenchmark.TOPIC_PREFIX + i;
-
-            SubscriptionOptions opts = SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).build();
-            subscriber.subscribe(ByteString.copyFromUtf8(topic), subId, opts);
-            subscriber.startDelivery(ByteString.copyFromUtf8(topic), subId, new MessageHandler() {
-
-                @Override
-                public void deliver(ByteString thisTopic, ByteString subscriberId, Message msg,
-                Callback<Void> callback, Object context) {
-                    logger.debug("Got message from src-region: {} with seq-id: {}",
-                        msg.getSrcRegion(), msg.getMsgId());
-
-                    String mapKey = topic + msg.getSrcRegion().toStringUtf8();
-                    Long lastSeqIdSeen = lastSeqIdSeenMap.get(mapKey);
-                    if (lastSeqIdSeen == null) {
-                        lastSeqIdSeen = (long) 0;
-                    }
-
-                    if (getSrcSeqId(msg) <= lastSeqIdSeen) {
-                        logger.info("Redelivery of message, src-region: " + msg.getSrcRegion() + "seq-id: "
-                                    + msg.getMsgId());
-                    } else {
-                        agg.ding(false);
-                    }
-
-                    callback.operationFinished(context, null);
-                }
-            });
-        }
-        System.out.println("Finished subscribing to topics and now waiting for messages to come in...");
-        // Wait till the benchmark test has completed
-        agg.queue.take();
-        System.out.println(agg.summarize(agg.earliest.get()));
-        return null;
-    }
-
-    long getSrcSeqId(Message msg) {
-        if (msg.getMsgId().getRemoteComponentsCount() == 0) {
-            return msg.getMsgId().getLocalComponent();
-        }
-
-        for (RegionSpecificSeqId rseqId : msg.getMsgId().getRemoteComponentsList()) {
-            if (rseqId.getRegion().equals(msg.getSrcRegion()))
-                return rseqId.getSeqId();
-        }
-
-        return msg.getMsgId().getLocalComponent();
-    }
-
-    void multiSub(String label, String topicPrefix, int start, final int npar, final int count)
-            throws InterruptedException {
-        long startTime = MathUtils.now();
-        ThroughputLatencyAggregator agg = new ThroughputLatencyAggregator(label, count / numPartitions, npar);
-        agg.startProgress();
-
-        int end = start + count;
-        for (int i = start; i < end; ++i) {
-            if (!HedwigBenchmark.amIResponsibleForTopic(i, partitionIndex, numPartitions)) {
-                continue;
-            }
-            SubscriptionOptions opts = SubscriptionOptions.newBuilder()
-                .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).build();
-            subscriber.asyncSubscribe(ByteString.copyFromUtf8(topicPrefix + i),
-                                      subId, opts,
-                                      new BenchmarkCallback(agg), null);
-        }
-        // Wait till the benchmark test has completed
-        agg.tpAgg.queue.take();
-        if (count > 1)
-            System.out.println(agg.summarize(startTime));
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkUtils.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkUtils.java b/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkUtils.java
deleted file mode 100644
index 3efe22d..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkUtils.java
+++ /dev/null
@@ -1,191 +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.client.benchmark;
-
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.bookkeeper.util.MathUtils;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.util.Callback;
-
-public class BenchmarkUtils {
-    static final Logger logger = LoggerFactory.getLogger(BenchmarkUtils.class);
-
-    public static double calcTp(final int count, long startTime) {
-        return 1000. * count / (MathUtils.now() - startTime);
-    }
-
-    /**
-     * Stats aggregator for callback (round-trip) operations. Measures both
-     * throughput and latency.
-     */
-    public static class ThroughputLatencyAggregator {
-        int numBuckets;
-        final ThroughputAggregator tpAgg;
-        final Semaphore outstanding;
-        final AtomicLong sum = new AtomicLong();
-
-        final AtomicLong[] latencyBuckets;
-
-        // bucket[i] is count of number of operations that took >= i ms and <
-        // (i+1) ms.
-
-        public ThroughputLatencyAggregator(String label, int count, int limit) throws InterruptedException {
-            numBuckets = Integer.getInteger("numBuckets", 101);
-            latencyBuckets = new AtomicLong[numBuckets];
-            tpAgg = new ThroughputAggregator(label, count);
-            outstanding = new Semaphore(limit);
-            for (int i = 0; i < numBuckets; i++) {
-                latencyBuckets[i] = new AtomicLong();
-            }
-        }
-
-        public void startProgress() {
-            tpAgg.startProgress();
-        }
-
-        public void reportLatency(long latency) {
-            sum.addAndGet(latency);
-
-            int bucketIndex;
-            if (latency >= numBuckets) {
-                bucketIndex = (int) numBuckets - 1;
-            } else {
-                bucketIndex = (int) latency;
-            }
-            latencyBuckets[bucketIndex].incrementAndGet();
-        }
-
-        private String getPercentile(double percentile) {
-            int numInliersNeeded = (int) (percentile / 100 * tpAgg.count);
-            int numInliersFound = 0;
-            for (int i = 0; i < numBuckets - 1; i++) {
-                numInliersFound += latencyBuckets[i].intValue();
-                if (numInliersFound > numInliersNeeded) {
-                    return i + "";
-                }
-            }
-            return " >= " + (numBuckets - 1);
-        }
-
-        public String summarize(long startTime) {
-            double percentile = Double.parseDouble(System.getProperty("percentile", "99.9"));
-            return tpAgg.summarize(startTime) + ", avg latency = " + sum.get() / tpAgg.count + ", " + percentile
-                   + "%ile latency = " + getPercentile(percentile);
-        }
-    }
-
-    /**
-     * Stats aggregator for non-callback (single-shot) operations. Measures just
-     * throughput.
-     */
-    public static class ThroughputAggregator {
-        final String label;
-        final int count;
-        final AtomicInteger done = new AtomicInteger();
-        final AtomicLong earliest = new AtomicLong();
-        final AtomicInteger numFailed = new AtomicInteger();
-        final Thread progressThread;
-        final LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
-
-        public ThroughputAggregator(final String label, final int count) {
-            this.label = label;
-            this.count = count;
-            if (count == 0)
-                queue.add(0);
-            if (Boolean.getBoolean("progress")) {
-                progressThread = new Thread(new Runnable() {
-                    @Override
-                    public void run() {
-                        try {
-                            for (int doneSnap = 0, prev = 0; doneSnap < count; prev = doneSnap, doneSnap = done.get()) {
-                                if (doneSnap > prev) {
-                                    System.out.println(label + " progress: " + doneSnap + " of " + count);
-                                }
-                                Thread.sleep(1000);
-                            }
-                        } catch (Exception ex) {
-                            throw new RuntimeException(ex);
-                        }
-                    }
-                    });
-            } else {
-                progressThread = null;
-            }
-        }
-
-        public void startProgress() {
-            if (progressThread != null) {
-                progressThread.start();
-            }
-        }
-
-        public void ding(boolean failed) {
-            int snapDone = done.incrementAndGet();
-            earliest.compareAndSet(0, MathUtils.now());
-            if (failed)
-                numFailed.incrementAndGet();
-            if (logger.isDebugEnabled())
-                logger.debug(label + " " + (failed ? "failed" : "succeeded") + ", done so far = " + snapDone);
-            if (snapDone == count) {
-                queue.add(numFailed.get());
-            }
-        }
-
-        public String summarize(long startTime) {
-            return "Finished " + label + ": count = " + done.get() + ", tput = " + calcTp(count, startTime)
-                   + " ops/s, numFailed = " + numFailed;
-        }
-    }
-
-    public static class BenchmarkCallback implements Callback<Void> {
-
-        final ThroughputLatencyAggregator agg;
-        final long startTime;
-
-        public BenchmarkCallback(ThroughputLatencyAggregator agg) throws InterruptedException {
-            this.agg = agg;
-            agg.outstanding.acquire();
-            // Must set the start time *after* taking acquiring on outstanding.
-            startTime = MathUtils.now();
-        }
-
-        private void finish(boolean failed) {
-            agg.reportLatency(MathUtils.now() - startTime);
-            agg.tpAgg.ding(failed);
-            agg.outstanding.release();
-        }
-
-        @Override
-        public void operationFinished(Object ctx, Void resultOfOperation) {
-            finish(false);
-        }
-
-        @Override
-        public void operationFailed(Object ctx, PubSubException exception) {
-            finish(true);
-        }
-    };
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkWorker.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkWorker.java b/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkWorker.java
deleted file mode 100644
index e7b15f2..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/BenchmarkWorker.java
+++ /dev/null
@@ -1,46 +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.client.benchmark;
-
-public class BenchmarkWorker {
-    int numTopics;
-    int numMessages;
-    int numRegions;
-    int startTopicLabel;
-    int partitionIndex;
-    int numPartitions;
-
-    public BenchmarkWorker(int numTopics, int numMessages, int numRegions,
-                           int startTopicLabel, int partitionIndex, int numPartitions) {
-        this.numTopics = numTopics;
-        this.numMessages = numMessages;
-        this.numRegions = numRegions;
-        this.startTopicLabel = startTopicLabel;
-        this.partitionIndex = partitionIndex;
-        this.numPartitions = numPartitions;
-
-        if (numMessages % (numTopics * numRegions) != 0) {
-            throw new RuntimeException("Number of messages not equally divisible among regions and topics");
-        }
-
-        if (numTopics % numPartitions != 0) {
-            throw new RuntimeException("Number of topics not equally divisible among partitions");
-        }
-
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java b/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java
deleted file mode 100644
index cc5e937..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java
+++ /dev/null
@@ -1,165 +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.client.benchmark;
-
-import java.io.File;
-import java.util.concurrent.Callable;
-
-import org.apache.commons.configuration.ConfigurationException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.jboss.netty.logging.InternalLoggerFactory;
-import org.jboss.netty.logging.Log4JLoggerFactory;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.util.HedwigSocketAddress;
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.client.HedwigClient;
-import org.apache.hedwig.client.api.Publisher;
-import org.apache.hedwig.client.api.Subscriber;
-
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.PosixParser;
-import org.apache.commons.cli.ParseException;
-
-public class HedwigBenchmark implements Callable<Void> {
-    protected static final Logger logger = LoggerFactory.getLogger(HedwigBenchmark.class);
-
-    static final String TOPIC_PREFIX = "topic";
-
-    private final HedwigClient client;
-    private final Publisher publisher;
-    private final Subscriber subscriber;
-    private final CommandLine cmd;
-
-    public HedwigBenchmark(ClientConfiguration cfg, CommandLine cmd) {
-        client = new HedwigClient(cfg);
-        publisher = client.getPublisher();
-        subscriber = client.getSubscriber();
-        this.cmd = cmd;
-    }
-
-    static boolean amIResponsibleForTopic(int topicNum, int partitionIndex, int numPartitions) {
-        return topicNum % numPartitions == partitionIndex;
-    }
-
-    @Override
-    public Void call() throws Exception {
-
-        //
-        // Parameters.
-        //
-
-        // What program to run: pub, sub (subscription benchmark), recv.
-        final String mode = cmd.getOptionValue("mode","");
-
-        // Number of requests to make (publishes or subscribes).
-        int numTopics = Integer.valueOf(cmd.getOptionValue("nTopics", "50"));
-        int numMessages = Integer.valueOf(cmd.getOptionValue("nMsgs", "1000"));
-        int numRegions = Integer.valueOf(cmd.getOptionValue("nRegions", "1"));
-        int startTopicLabel = Integer.valueOf(cmd.getOptionValue("startTopicLabel", "0"));
-        int partitionIndex = Integer.valueOf(cmd.getOptionValue("partitionIndex", "0"));
-        int numPartitions = Integer.valueOf(cmd.getOptionValue("nPartitions", "1"));
-
-        int replicaIndex = Integer.valueOf(cmd.getOptionValue("replicaIndex", "0"));
-
-        int rate = Integer.valueOf(cmd.getOptionValue("rate", "0"));
-        int nParallel = Integer.valueOf(cmd.getOptionValue("npar", "100"));
-        int msgSize = Integer.valueOf(cmd.getOptionValue("msgSize", "1024"));
-
-        // Number of warmup subscriptions to make.
-        final int nWarmups = Integer.valueOf(cmd.getOptionValue("nwarmups", "1000"));
-
-        if (mode.equals("sub")) {
-            BenchmarkSubscriber benchmarkSub = new BenchmarkSubscriber(numTopics, 0, 1, startTopicLabel, 0, 1,
-                    subscriber, ByteString.copyFromUtf8("mySub"));
-
-            benchmarkSub.warmup(nWarmups);
-            benchmarkSub.call();
-
-        } else if (mode.equals("recv")) {
-
-            BenchmarkSubscriber benchmarkSub = new BenchmarkSubscriber(numTopics, numMessages, numRegions,
-                    startTopicLabel, partitionIndex, numPartitions, subscriber, ByteString.copyFromUtf8("sub-"
-                            + replicaIndex));
-
-            benchmarkSub.call();
-
-        } else if (mode.equals("pub")) {
-            // Offered load in msgs/second.
-            BenchmarkPublisher benchmarkPub = new BenchmarkPublisher(numTopics, numMessages, numRegions,
-                    startTopicLabel, partitionIndex, numPartitions, publisher, subscriber, msgSize, nParallel, rate);
-            benchmarkPub.warmup(nWarmups);
-            benchmarkPub.call();
-
-        } else {
-            throw new Exception("unknown mode: " + mode);
-        }
-
-        return null;
-    }
-
-    public static void main(String[] args) throws Exception {
-        Options options = new Options();
-        options.addOption("mode", true, "sub, recv, or pub");
-        options.addOption("nTopics", true, "Number of topics, default 50");
-        options.addOption("nMsgs", true, "Number of messages, default 1000");
-        options.addOption("nRegions", true, "Number of regsions, default 1");
-        options.addOption("startTopicLabel", true,
-                          "Prefix of topic labels. Must be numeric. Default 0");
-        options.addOption("partitionIndex", true, "If partitioning, the partition index for this client");
-        options.addOption("nPartitions", true, "Number of partitions, default 1");
-        options.addOption("replicaIndex", true, "default 0");
-        options.addOption("rate", true, "default 0");
-        options.addOption("npar", true, "default 100");
-        options.addOption("msgSize", true, "Size of messages, default 1024");
-        options.addOption("nwarmups", true, "Number of warmup messages, default 1000");
-        options.addOption("defaultHub", true, "Default hedwig hub to connect to, default localhost:4080");
-
-        CommandLineParser parser = new PosixParser();
-        final CommandLine cmd = parser.parse(options, args);
-
-        if (cmd.hasOption("help")) {
-            HelpFormatter formatter = new HelpFormatter();
-            formatter.printHelp("HedwigBenchmark <options>", options);
-            System.exit(-1);
-        }
-
-        ClientConfiguration cfg = new ClientConfiguration() {
-                public HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
-                    return new HedwigSocketAddress(cmd.getOptionValue("defaultHub",
-                                                                      "localhost:4080"));
-                }
-
-                public boolean isSSLEnabled() {
-                    return false;
-                }
-            };
-
-        InternalLoggerFactory.setDefaultFactory(new Log4JLoggerFactory());
-
-        HedwigBenchmark app = new HedwigBenchmark(cfg, cmd);
-        app.call();
-        System.exit(0);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java b/hedwig-client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java
deleted file mode 100644
index 836a9d2..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java
+++ /dev/null
@@ -1,171 +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.client.conf;
-
-import java.net.InetSocketAddress;
-
-import org.apache.commons.configuration.ConfigurationException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.hedwig.conf.AbstractConfiguration;
-import org.apache.hedwig.util.HedwigSocketAddress;
-
-public class ClientConfiguration extends AbstractConfiguration {
-    private static final Logger logger = LoggerFactory.getLogger(ClientConfiguration.class);
-
-    // Protected member variables for configuration parameter names
-    protected static final String DEFAULT_SERVER_HOST = "default_server_host";
-    protected static final String MAX_MESSAGE_SIZE = "max_message_size";
-    protected static final String MAX_SERVER_REDIRECTS = "max_server_redirects";
-    protected static final String AUTO_SEND_CONSUME_MESSAGE_ENABLED = "auto_send_consume_message_enabled";
-    protected static final String CONSUMED_MESSAGES_BUFFER_SIZE = "consumed_messages_buffer_size";
-    protected static final String MESSAGE_CONSUME_RETRY_WAIT_TIME = "message_consume_retry_wait_time";
-    protected static final String SUBSCRIBE_RECONNECT_RETRY_WAIT_TIME = "subscribe_reconnect_retry_wait_time";
-    protected static final String MAX_OUTSTANDING_MESSAGES = "max_outstanding_messages";
-    protected static final String SERVER_ACK_RESPONSE_TIMEOUT = "server_ack_response_timeout";
-    protected static final String TIMEOUT_THREAD_RUN_INTERVAL = "timeout_thread_run_interval";
-    protected static final String SSL_ENABLED = "ssl_enabled";
-    protected static final String SUBSCRIPTION_MESSAGE_BOUND = "subscription_message_bound";
-    protected static final String SUBSCRIPTION_CHANNEL_SHARING_ENABLED = "subscription_channel_sharing_enabled";
-
-    // Singletons we want to instantiate only once per ClientConfiguration
-    protected HedwigSocketAddress myDefaultServerAddress = null;
-
-    // Getters for the various Client Configuration parameters.
-    // This should point to the default server host, or the VIP fronting all of
-    // the server hubs. This will return the HedwigSocketAddress which
-    // encapsulates both the regular and SSL port connection to the server host.
-    protected HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
-        if (myDefaultServerAddress == null)
-            myDefaultServerAddress = new HedwigSocketAddress(conf.getString(DEFAULT_SERVER_HOST, "localhost:4080:9876"));
-        return myDefaultServerAddress;
-    }
-
-    // This will get the default server InetSocketAddress based on if SSL is
-    // enabled or not.
-    public InetSocketAddress getDefaultServerHost() {
-        if (isSSLEnabled())
-            return getDefaultServerHedwigSocketAddress().getSSLSocketAddress();
-        else
-            return getDefaultServerHedwigSocketAddress().getSocketAddress();
-    }
-
-    public int getMaximumMessageSize() {
-        return conf.getInt(MAX_MESSAGE_SIZE, 2 * 1024 * 1024);
-    }
-
-    // This parameter is for setting the maximum number of server redirects to
-    // allow before we consider it as an error condition. This is to stop
-    // infinite redirect loops in case there is a problem with the hub servers
-    // topic mastership.
-    public int getMaximumServerRedirects() {
-        return conf.getInt(MAX_SERVER_REDIRECTS, 2);
-    }
-
-    // This parameter is a boolean flag indicating if the client library should
-    // automatically send the consume message to the server based on the
-    // configured amount of messages consumed by the client app. The client app
-    // could choose to override this behavior and instead, manually send the
-    // consume message to the server via the client library using its own
-    // logic and policy.
-    public boolean isAutoSendConsumeMessageEnabled() {
-        return conf.getBoolean(AUTO_SEND_CONSUME_MESSAGE_ENABLED, true);
-    }
-
-    // This parameter is to set how many consumed messages we'll buffer up
-    // before we send the Consume message to the server indicating that all
-    // of the messages up to that point have been successfully consumed by
-    // the client.
-    public int getConsumedMessagesBufferSize() {
-        return conf.getInt(CONSUMED_MESSAGES_BUFFER_SIZE, 5);
-    }
-
-    // This parameter is used to determine how long we wait before retrying the
-    // client app's MessageHandler to consume a subscribed messages sent to us
-    // from the server. The time to wait is in milliseconds.
-    public long getMessageConsumeRetryWaitTime() {
-        return conf.getLong(MESSAGE_CONSUME_RETRY_WAIT_TIME, 10000);
-    }
-
-    // This parameter is used to determine how long we wait before retrying the
-    // Subscribe Reconnect request. This is done when the connection to a server
-    // disconnects and we attempt to connect to it. We'll keep on trying but
-    // in case the server(s) is down for a longer time, we want to throttle
-    // how often we do the subscribe reconnect request. The time to wait is in
-    // milliseconds.
-    public long getSubscribeReconnectRetryWaitTime() {
-        return conf.getLong(SUBSCRIBE_RECONNECT_RETRY_WAIT_TIME, 10000);
-    }
-
-    // This parameter is for setting the maximum number of outstanding messages
-    // the client app can be consuming at a time for topic subscription before
-    // we throttle things and stop reading from the Netty Channel.
-    public int getMaximumOutstandingMessages() {
-        return conf.getInt(MAX_OUTSTANDING_MESSAGES, 10);
-    }
-
-    // This parameter is used to determine how long we wait (in milliseconds)
-    // before we time out outstanding PubSubRequests that were written to the
-    // server successfully but haven't yet received the ack response.
-    public long getServerAckResponseTimeout() {
-        return conf.getLong(SERVER_ACK_RESPONSE_TIMEOUT, 30000);
-    }
-
-    // This parameter is used to determine how often we run the server ack
-    // response timeout cleaner thread (in milliseconds).
-    public long getTimeoutThreadRunInterval() {
-        return conf.getLong(TIMEOUT_THREAD_RUN_INTERVAL, 60000);
-    }
-
-    // This parameter is a boolean flag indicating if communication with the
-    // server should be done via SSL for encryption. This is needed for
-    // cross-colo hub clients listening to non-local servers.
-    public boolean isSSLEnabled() {
-        return conf.getBoolean(SSL_ENABLED, false);
-    }
-
-    /**
-     * This parameter is a boolean flag indicating if multiplexing subscription
-     * channels.
-     */
-    public boolean isSubscriptionChannelSharingEnabled() {
-        return conf.getBoolean(SUBSCRIPTION_CHANNEL_SHARING_ENABLED, false);
-    }
-
-    /**
-     * The maximum number of messages the hub will queue for subscriptions
-     * created using this configuration. The hub will always queue the most
-     * recent messages. If there are enough publishes to the topic to hit
-     * the bound, then the oldest messages are dropped from the queue.
-     *
-     * A bound of 0 disabled the bound completely. This is the default.
-     */
-    public int getSubscriptionMessageBound() {
-        return conf.getInt(SUBSCRIPTION_MESSAGE_BOUND, 0);
-    }
-
-    // Validate that the configuration properties are valid.
-    public void validate() throws ConfigurationException {
-        if (isSSLEnabled() && getDefaultServerHedwigSocketAddress().getSSLSocketAddress() == null) {
-            throw new ConfigurationException("SSL is enabled but a default server SSL port not given!");
-        }
-        // Add other validation checks here
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java b/hedwig-client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java
deleted file mode 100644
index 346d74b..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java
+++ /dev/null
@@ -1,56 +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.client.data;
-
-import org.apache.hedwig.client.data.TopicSubscriber;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-
-/**
- * Wrapper class to store all of the data points needed to encapsulate Message
- * Consumption in the Subscribe flow for consuming a message sent from the
- * server for a given TopicSubscriber. This will be used as the Context in the
- * VoidCallback for the MessageHandlers once they've completed consuming the
- * message.
- *
- */
-public class MessageConsumeData {
-
-    // Member variables
-    public final TopicSubscriber topicSubscriber;
-    // This is the Message sent from the server for Subscribes for consumption
-    // by the client.
-    public final Message msg;
-
-    // Constructor
-    public MessageConsumeData(final TopicSubscriber topicSubscriber, final Message msg) {
-        this.topicSubscriber = topicSubscriber;
-        this.msg = msg;
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder sb = new StringBuilder();
-        if (topicSubscriber != null) {
-            sb.append("Subscription: ").append(topicSubscriber);
-        }
-        if (msg != null) {
-            sb.append(PubSubData.COMMA).append("Message: ").append(msg);
-        }
-        return sb.toString();
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/data/PubSubData.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/data/PubSubData.java b/hedwig-client/src/main/java/org/apache/hedwig/client/data/PubSubData.java
deleted file mode 100644
index 63547a0..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/data/PubSubData.java
+++ /dev/null
@@ -1,179 +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.client.data;
-
-import java.util.List;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.client.netty.HChannel;
-import org.apache.hedwig.protocol.PubSubProtocol;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.protocol.PubSubProtocol.OperationType;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions;
-import org.apache.hedwig.util.Callback;
-
-/**
- * Wrapper class to store all of the data points needed to encapsulate all
- * PubSub type of request operations the client will do. This includes knowing
- * all of the information needed if we need to redo the publish/subscribe
- * request in case of a server redirect. This will be used for all sync/async
- * calls, and for all the known types of request messages to send to the server
- * hubs: Publish, Subscribe, Unsubscribe, and Consume.
- *
- */
-public class PubSubData {
-    // Static string constants
-    protected static final String COMMA = ", ";
-
-    // Member variables needed during object construction time.
-    public final ByteString topic;
-    public final Message msg;
-    public final ByteString subscriberId;
-    // Enum to indicate what type of operation this PubSub request data object
-    // is for.
-    public final OperationType operationType;
-    // Options for the subscription
-    public final SubscriptionOptions options;
-
-    // These two variables are not final since we might override them
-    // in the case of a Subscribe reconnect.
-    private Callback<PubSubProtocol.ResponseBody> callback;
-
-    public Object context;
-
-    // Member variables used after object has been constructed.
-    // List of all servers we've sent the PubSubRequest to successfully.
-    // This is to keep track of redirected servers that responded back to us.
-    public List<ByteString> triedServers;
-    // List of all servers that we've tried to connect or write to but
-    // was unsuccessful. We'll retry sending the PubSubRequest but will
-    // quit if we're trying to connect or write to a server that we've
-    // attempted to previously.
-    public List<ByteString> connectFailedServers;
-    public List<ByteString> writeFailedServers;
-    // Boolean to the hub server indicating if it should claim ownership
-    // of the topic the PubSubRequest is for. This is mainly used after
-    // a server redirect. Defaults to false.
-    public boolean shouldClaim = false;
-    // TxnID for the PubSubData if it was sent as a PubSubRequest to the hub
-    // server. This is used in the WriteCallback in case of failure. We want
-    // to remove it from the ResponseHandler.txn2PubSubData map since the
-    // failed PubSubRequest will not get an ack response from the server.
-    // This is set later in the PubSub flows only when we write the actual
-    // request. Therefore it is not an argument in the constructor.
-    public long txnId;
-    // Time in milliseconds using the System.currentTimeMillis() call when the
-    // PubSubRequest was written on the netty Channel to the server.
-    public long requestWriteTime;
-    // For synchronous calls, this variable is used to know when the background
-    // async process for it has completed, set in the VoidCallback.
-    public boolean isDone = false;
-    // Record the original channel for a resubscribe request
-    private HChannel origChannel = null;
-
-    // Constructor for all types of PubSub request data to send to the server
-    public PubSubData(final ByteString topic, final Message msg, final ByteString subscriberId,
-                      final OperationType operationType, final SubscriptionOptions options,
-                      final Callback<PubSubProtocol.ResponseBody> callback,
-                      final Object context) {
-        this.topic = topic;
-        this.msg = msg;
-        this.subscriberId = subscriberId;
-        this.operationType = operationType;
-        this.options = options;
-        this.callback = callback;
-        this.context = context;
-    }
-
-    public void setCallback(Callback<PubSubProtocol.ResponseBody> callback) {
-        this.callback = callback;
-    }
-
-    public Callback<PubSubProtocol.ResponseBody> getCallback() {
-        return callback;
-    }
-
-    public void operationFinishedToCallback(Object context, PubSubProtocol.ResponseBody response){
-        callback.operationFinished(context, response);
-    }
-
-    public boolean isResubscribeRequest() {
-        return null != origChannel;
-    }
-
-    public HChannel getOriginalChannelForResubscribe() {
-        return origChannel;
-    }
-
-    public void setOriginalChannelForResubscribe(HChannel channel) {
-        this.origChannel = channel;
-    }
-
-    // Clear all of the stored servers we've contacted or attempted to in this
-    // request.
-    public void clearServersList() {
-        if (triedServers != null)
-            triedServers.clear();
-        if (connectFailedServers != null)
-            connectFailedServers.clear();
-        if (writeFailedServers != null)
-            writeFailedServers.clear();
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder sb = new StringBuilder();
-        if (topic != null)
-            sb.append("Topic: " + topic.toStringUtf8());
-        if (msg != null)
-            sb.append(COMMA).append("Message: " + msg);
-        if (subscriberId != null)
-            sb.append(COMMA).append("SubscriberId: " + subscriberId.toStringUtf8());
-        if (operationType != null)
-            sb.append(COMMA).append("Operation Type: " + operationType.toString());
-        if (options != null)
-            sb.append(COMMA).append("Create Or Attach: " + options.getCreateOrAttach().toString())
-                .append(COMMA).append("Message Bound: " + options.getMessageBound());
-        if (triedServers != null && triedServers.size() > 0) {
-            sb.append(COMMA).append("Tried Servers: ");
-            for (ByteString triedServer : triedServers) {
-                sb.append(triedServer.toStringUtf8()).append(COMMA);
-            }
-        }
-        if (connectFailedServers != null && connectFailedServers.size() > 0) {
-            sb.append(COMMA).append("Connect Failed Servers: ");
-            for (ByteString connectFailedServer : connectFailedServers) {
-                sb.append(connectFailedServer.toStringUtf8()).append(COMMA);
-            }
-        }
-        if (writeFailedServers != null && writeFailedServers.size() > 0) {
-            sb.append(COMMA).append("Write Failed Servers: ");
-            for (ByteString writeFailedServer : writeFailedServers) {
-                sb.append(writeFailedServer.toStringUtf8()).append(COMMA);
-            }
-        }
-        sb.append(COMMA).append("Should Claim: " + shouldClaim);
-        if (txnId != 0)
-            sb.append(COMMA).append("TxnID: " + txnId);
-        if (requestWriteTime != 0)
-            sb.append(COMMA).append("Request Write Time: " + requestWriteTime);
-        sb.append(COMMA).append("Is Done: " + isDone);
-        return sb.toString();
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java b/hedwig-client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java
deleted file mode 100644
index 064cec1..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java
+++ /dev/null
@@ -1,74 +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.client.data;
-
-import org.apache.commons.lang.builder.HashCodeBuilder;
-
-import com.google.protobuf.ByteString;
-
-/**
- * Wrapper class object for the Topic + SubscriberId combination. Since the
- * Subscribe flows always use the Topic + SubscriberId as the logical entity,
- * we'll create a simple class to encapsulate that.
- *
- */
-public class TopicSubscriber {
-    private final ByteString topic;
-    private final ByteString subscriberId;
-    private final int hashCode;
-
-    public TopicSubscriber(final ByteString topic, final ByteString subscriberId) {
-        this.topic = topic;
-        this.subscriberId = subscriberId;
-        hashCode = new HashCodeBuilder().append(topic).append(subscriberId).toHashCode();
-    }
-
-    @Override
-    public boolean equals(final Object o) {
-        if (o == this)
-            return true;
-        if (!(o instanceof TopicSubscriber))
-            return false;
-        final TopicSubscriber obj = (TopicSubscriber) o;
-        return topic.equals(obj.topic) && subscriberId.equals(obj.subscriberId);
-    }
-
-    @Override
-    public int hashCode() {
-        return hashCode;
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder sb = new StringBuilder();
-        if (topic != null)
-            sb.append("Topic: " + topic.toStringUtf8());
-        if (subscriberId != null)
-            sb.append(PubSubData.COMMA).append("SubscriberId: " + subscriberId.toStringUtf8());
-        return sb.toString();
-    }
-
-    public ByteString getTopic() {
-        return topic;
-    }
-
-    public ByteString getSubscriberId() {
-        return subscriberId;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/AlreadyStartDeliveryException.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/AlreadyStartDeliveryException.java b/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/AlreadyStartDeliveryException.java
deleted file mode 100644
index 5f468e6..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/AlreadyStartDeliveryException.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.client.exceptions;
-
-/**
- * This is a Hedwig client side exception when the local client wants to
- * startDelivery using another message handler before stopping previous one.
- */
-public class AlreadyStartDeliveryException extends Exception {
-
-    private static final long serialVersionUID = 873259807218723524L;
-
-    public AlreadyStartDeliveryException(String message) {
-        super(message);
-    }
-
-    public AlreadyStartDeliveryException(String message, Throwable t) {
-        super(message, t);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java b/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java
deleted file mode 100644
index 3e54356..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java
+++ /dev/null
@@ -1,37 +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.client.exceptions;
-
-/**
- * This is a Hedwig client side exception when the local client wants to do
- * subscribe type of operations. Currently, to distinguish between local and hub
- * subscribers, the subscriberId will have a specific format.
- */
-public class InvalidSubscriberIdException extends Exception {
-
-    private static final long serialVersionUID = 873259807218723523L;
-
-    public InvalidSubscriberIdException(String message) {
-        super(message);
-    }
-
-    public InvalidSubscriberIdException(String message, Throwable t) {
-        super(message, t);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java b/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java
deleted file mode 100644
index 22b44b1..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java
+++ /dev/null
@@ -1,34 +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.client.exceptions;
-
-/**
- * This is a Hedwig client side exception thrown when it can't get the response
- * handler from the channel pipeline responsible for a PubSubRequest.
- */
-public class NoResponseHandlerException extends Exception {
-    private static final long serialVersionUID = 1L;
-
-    public NoResponseHandlerException(String message) {
-        super(message);
-    }
-
-    public NoResponseHandlerException(String message, Throwable t) {
-        super(message, t);
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ResubscribeException.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ResubscribeException.java b/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ResubscribeException.java
deleted file mode 100644
index c9aeb38..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ResubscribeException.java
+++ /dev/null
@@ -1,34 +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.client.exceptions;
-
-/**
- * This is a Hedwig client side exception when the client failed to resubscribe
- * when topic moved or subscription is closed.
- */
-public class ResubscribeException extends Exception {
-
-    public ResubscribeException(String message) {
-        super(message);
-    }
-
-    public ResubscribeException(String message, Throwable t) {
-        super(message, t);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java b/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java
deleted file mode 100644
index da6d4e7..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java
+++ /dev/null
@@ -1,38 +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.client.exceptions;
-
-/**
- * This is a Hedwig client side exception when the PubSubRequest is being
- * redirected to a server where the request has already been sent to previously.
- * To avoid having a cyclical redirect loop, this condition is checked for
- * and this exception will be thrown to the client caller.
- */
-public class ServerRedirectLoopException extends Exception {
-
-    private static final long serialVersionUID = 98723508723152897L;
-
-    public ServerRedirectLoopException(String message) {
-        super(message);
-    }
-
-    public ServerRedirectLoopException(String message, Throwable t) {
-        super(message, t);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java b/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java
deleted file mode 100644
index 4a3c99f..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java
+++ /dev/null
@@ -1,39 +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.client.exceptions;
-
-/**
- * This is a Hedwig client side exception when there have been too many server
- * redirects during a publish/subscribe call. We only allow a certain number of
- * server redirects to find the topic master. If we have exceeded this
- * configured amount, the publish/subscribe will fail with this exception.
- *
- */
-public class TooManyServerRedirectsException extends Exception {
-
-    private static final long serialVersionUID = 2341192937965635310L;
-
-    public TooManyServerRedirectsException(String message) {
-        super(message);
-    }
-
-    public TooManyServerRedirectsException(String message, Throwable t) {
-        super(message, t);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/AbstractResponseHandler.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/AbstractResponseHandler.java b/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/AbstractResponseHandler.java
deleted file mode 100644
index b8a6787..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/AbstractResponseHandler.java
+++ /dev/null
@@ -1,156 +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.client.handlers;
-
-import java.net.InetSocketAddress;
-import java.util.LinkedList;
-
-import com.google.protobuf.ByteString;
-
-import org.jboss.netty.channel.Channel;
-
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.client.data.PubSubData;
-import org.apache.hedwig.client.netty.HChannelManager;
-import org.apache.hedwig.client.exceptions.ServerRedirectLoopException;
-import org.apache.hedwig.client.exceptions.TooManyServerRedirectsException;
-import org.apache.hedwig.client.netty.NetUtils;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
-import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
-import org.apache.hedwig.util.HedwigSocketAddress;
-import static org.apache.hedwig.util.VarArgs.va;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public abstract class AbstractResponseHandler {
-
-    private final static Logger logger = LoggerFactory.getLogger(AbstractResponseHandler.class);
-
-    protected final ClientConfiguration cfg;
-    protected final HChannelManager channelManager;
-
-    protected AbstractResponseHandler(ClientConfiguration cfg,
-                                      HChannelManager channelManager) {
-        this.cfg = cfg;
-        this.channelManager = channelManager;
-    }
-
-    /**
-     * Logic to handle received response.
-     *
-     * @param response
-     *            PubSubResponse received from hub server.
-     * @param pubSubData
-     *            PubSubData for the pub/sub request.
-     * @param channel
-     *            Channel we used to make the request.
-     */
-    public abstract void handleResponse(PubSubResponse response, PubSubData pubSubData,
-                                        Channel channel) throws Exception;
-
-    /**
-     * Logic to repost a PubSubRequest when the server responds with a redirect
-     * indicating they are not the topic master.
-     *
-     * @param response
-     *            PubSubResponse from the server for the redirect
-     * @param pubSubData
-     *            PubSubData for the original PubSubRequest made
-     * @param channel
-     *            Channel Channel we used to make the original PubSubRequest
-     * @throws Exception
-     *             Throws an exception if there was an error in doing the
-     *             redirect repost of the PubSubRequest
-     */
-    protected void handleRedirectResponse(PubSubResponse response, PubSubData pubSubData,
-                                          Channel channel)
-            throws Exception {
-        if (logger.isDebugEnabled()) {
-            logger.debug("Handling a redirect from host: {}, response: {}, pubSubData: {}",
-                         va(NetUtils.getHostFromChannel(channel), response, pubSubData));
-        }
-        // In this case, the PubSub request was done to a server that is not
-        // responsible for the topic. First make sure that we haven't
-        // exceeded the maximum number of server redirects.
-        int curNumServerRedirects = (pubSubData.triedServers == null) ? 0 : pubSubData.triedServers.size();
-        if (curNumServerRedirects >= cfg.getMaximumServerRedirects()) {
-            // We've already exceeded the maximum number of server redirects
-            // so consider this as an error condition for the client.
-            // Invoke the operationFailed callback and just return.
-            logger.debug("Exceeded the number of server redirects ({}) so error out.",
-                         curNumServerRedirects);
-            PubSubException exception = new ServiceDownException(
-                new TooManyServerRedirectsException("Already reached max number of redirects: "
-                                                    + curNumServerRedirects));
-            pubSubData.getCallback().operationFailed(pubSubData.context, exception);
-            return;
-        }
-
-        // We will redirect and try to connect to the correct server
-        // stored in the StatusMsg of the response. First store the
-        // server that we sent the PubSub request to for the topic.
-        ByteString triedServer = ByteString.copyFromUtf8(HedwigSocketAddress.sockAddrStr(
-                                                         NetUtils.getHostFromChannel(channel)));
-        if (pubSubData.triedServers == null) {
-            pubSubData.triedServers = new LinkedList<ByteString>();
-        }
-        pubSubData.shouldClaim = true;
-        pubSubData.triedServers.add(triedServer);
-
-        // Now get the redirected server host (expected format is
-        // Hostname:Port:SSLPort) from the server's response message. If one is
-        // not given for some reason, then redirect to the default server
-        // host/VIP to repost the request.
-        String statusMsg = response.getStatusMsg();
-        InetSocketAddress redirectedHost;
-        boolean redirectToDefaultServer;
-        if (statusMsg != null && statusMsg.length() > 0) {
-            if (cfg.isSSLEnabled()) {
-                redirectedHost = new HedwigSocketAddress(statusMsg).getSSLSocketAddress();
-            } else {
-                redirectedHost = new HedwigSocketAddress(statusMsg).getSocketAddress();
-            }
-            redirectToDefaultServer = false;
-        } else {
-            redirectedHost = cfg.getDefaultServerHost();
-            redirectToDefaultServer = true;
-        }
-
-        // Make sure the redirected server is not one we've already attempted
-        // already before in this PubSub request.
-        if (pubSubData.triedServers.contains(ByteString.copyFromUtf8(HedwigSocketAddress.sockAddrStr(redirectedHost)))) {
-            logger.error("We've already sent this PubSubRequest before to redirectedHost: {}, pubSubData: {}",
-                         va(redirectedHost, pubSubData));
-            PubSubException exception = new ServiceDownException(
-                new ServerRedirectLoopException("Already made the request before to redirected host: "
-                                                + redirectedHost));
-            pubSubData.getCallback().operationFailed(pubSubData.context, exception);
-            return;
-        }
-
-        // submit the pub/sub request to redirected host
-        if (redirectToDefaultServer) {
-            channelManager.submitOpToDefaultServer(pubSubData);
-        } else {
-            channelManager.redirectToHost(pubSubData, redirectedHost);
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/CloseSubscriptionResponseHandler.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/CloseSubscriptionResponseHandler.java b/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/CloseSubscriptionResponseHandler.java
deleted file mode 100644
index c5b58a0..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/CloseSubscriptionResponseHandler.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.client.handlers;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.jboss.netty.channel.Channel;
-
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.client.data.PubSubData;
-import org.apache.hedwig.client.netty.HChannelManager;
-import org.apache.hedwig.exceptions.PubSubException.ClientNotSubscribedException;
-import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
-import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
-
-public class CloseSubscriptionResponseHandler extends AbstractResponseHandler {
-
-    private final static Logger logger = LoggerFactory.getLogger(CloseSubscriptionResponseHandler.class);
-
-    public CloseSubscriptionResponseHandler(ClientConfiguration cfg,
-                                            HChannelManager channelManager) {
-        super(cfg, channelManager);
-    }
-
-    @Override
-    public void handleResponse(final PubSubResponse response, final PubSubData pubSubData,
-                               final Channel channel)
-            throws Exception {
-        switch (response.getStatusCode()) {
-        case SUCCESS:
-            pubSubData.getCallback().operationFinished(pubSubData.context, null);
-            break;
-        case CLIENT_NOT_SUBSCRIBED:
-            // For closesubscription requests, the server says that the client was
-            // never subscribed to the topic.
-            pubSubData.getCallback().operationFailed(pubSubData.context, new ClientNotSubscribedException(
-                                                    "Client was never subscribed to topic: " +
-                                                        pubSubData.topic.toStringUtf8() + ", subscriberId: " +
-                                                        pubSubData.subscriberId.toStringUtf8()));
-            break;
-        case SERVICE_DOWN:
-            // Response was service down failure so just invoke the callback's
-            // operationFailed method.
-            pubSubData.getCallback().operationFailed(pubSubData.context, new ServiceDownException(
-                                                    "Server responded with a SERVICE_DOWN status"));
-            break;
-        case NOT_RESPONSIBLE_FOR_TOPIC:
-            // Redirect response so we'll need to repost the original
-            // Unsubscribe Request
-            handleRedirectResponse(response, pubSubData, channel);
-            break;
-        default:
-            // Consider all other status codes as errors, operation failed
-            // cases.
-            logger.error("Unexpected error response from server for PubSubResponse: " + response);
-            pubSubData.getCallback().operationFailed(pubSubData.context, new ServiceDownException(
-                                                    "Server responded with a status code of: " +
-                                                        response.getStatusCode()));
-            break;
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java b/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java
deleted file mode 100644
index 32517b0..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java
+++ /dev/null
@@ -1,117 +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.client.handlers;
-
-import java.util.TimerTask;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.client.data.MessageConsumeData;
-import org.apache.hedwig.client.netty.HChannelManager;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.protoextensions.MessageIdUtils;
-import org.apache.hedwig.util.Callback;
-import static org.apache.hedwig.util.VarArgs.va;
-
-/**
- * This is the Callback used by the MessageHandlers on the client app when
- * they've finished consuming a subscription message sent from the server
- * asynchronously. This callback back to the client libs will be stateless so we
- * can use a singleton for the class. The object context used should be the
- * MessageConsumeData type. That will contain all of the information needed to
- * call the message consume logic in the client lib HChannelHandler.
- *
- */
-public class MessageConsumeCallback implements Callback<Void> {
-
-    private static final Logger logger = LoggerFactory.getLogger(MessageConsumeCallback.class);
-
-    private final HChannelManager channelManager;
-    private final long consumeRetryWaitTime;
-
-    public MessageConsumeCallback(ClientConfiguration cfg,
-                                  HChannelManager channelManager) {
-        this.channelManager = channelManager;
-        this.consumeRetryWaitTime =
-            cfg.getMessageConsumeRetryWaitTime();
-    }
-
-    class MessageConsumeRetryTask extends TimerTask {
-        private final MessageConsumeData messageConsumeData;
-
-        public MessageConsumeRetryTask(MessageConsumeData messageConsumeData) {
-            this.messageConsumeData = messageConsumeData;
-        }
-
-        @Override
-        public void run() {
-            // Try to consume the message again
-            SubscribeResponseHandler subscribeHChannelHandler =
-                channelManager.getSubscribeResponseHandler(messageConsumeData.topicSubscriber);
-            if (null == subscribeHChannelHandler ||
-                !subscribeHChannelHandler.hasSubscription(messageConsumeData.topicSubscriber)) {
-                logger.warn("No subscription {} found to retry delivering message {}.",
-                            va(messageConsumeData.topicSubscriber,
-                               MessageIdUtils.msgIdToReadableString(messageConsumeData.msg.getMsgId())));
-                return;
-            }
-
-            subscribeHChannelHandler.asyncMessageDeliver(messageConsumeData.topicSubscriber,
-                                                         messageConsumeData.msg);
-        }
-    }
-
-    public void operationFinished(Object ctx, Void resultOfOperation) {
-        MessageConsumeData messageConsumeData = (MessageConsumeData) ctx;
-
-        SubscribeResponseHandler subscribeHChannelHandler =
-            channelManager.getSubscribeResponseHandler(messageConsumeData.topicSubscriber);
-        if (null == subscribeHChannelHandler ||
-            !subscribeHChannelHandler.hasSubscription(messageConsumeData.topicSubscriber)) {
-            logger.warn("No subscription {} found to consume message {}.",
-                        va(messageConsumeData.topicSubscriber,
-                           MessageIdUtils.msgIdToReadableString(messageConsumeData.msg.getMsgId())));
-            return;
-        }
-
-        // Message has been successfully consumed by the client app so callback
-        // to the HChannelHandler indicating that the message is consumed.
-        subscribeHChannelHandler.messageConsumed(messageConsumeData.topicSubscriber,
-                                                 messageConsumeData.msg);
-    }
-
-    public void operationFailed(Object ctx, PubSubException exception) {
-        // Message has NOT been successfully consumed by the client app so
-        // callback to the HChannelHandler to try the async MessageHandler
-        // Consume logic again.
-        MessageConsumeData messageConsumeData = (MessageConsumeData) ctx;
-        logger.error("Message was not consumed successfully by client MessageHandler: {}",
-                     messageConsumeData);
-
-        // Sleep a pre-configured amount of time (in milliseconds) before we
-        // do the retry. In the future, we can have more dynamic logic on
-        // what duration to sleep based on how many times we've retried, or
-        // perhaps what the last amount of time we slept was. We could stick
-        // some of this meta-data into the MessageConsumeData when we retry.
-        channelManager.schedule(new MessageConsumeRetryTask(messageConsumeData),
-                                consumeRetryWaitTime);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java b/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java
deleted file mode 100644
index 3f5fbcd..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java
+++ /dev/null
@@ -1,94 +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.client.handlers;
-
-import org.apache.hedwig.protocol.PubSubProtocol;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.hedwig.client.data.PubSubData;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.util.Callback;
-
-/**
- * This class is used when we are doing synchronous type of operations. All
- * underlying client ops in Hedwig are async so this is just a way to make the
- * async calls synchronous.
- *
- */
-public class PubSubCallback implements Callback<PubSubProtocol.ResponseBody> {
-
-    private static final Logger logger = LoggerFactory.getLogger(PubSubCallback.class);
-
-    // Private member variables
-    private final PubSubData pubSubData;
-    // Boolean indicator to see if the sync PubSub call was successful or not.
-    private boolean isCallSuccessful;
-    // For sync callbacks, we'd like to know what the PubSubException is thrown
-    // on failure. This is so we can have a handle to the exception and rethrow
-    // it later.
-    private PubSubException failureException;
-
-    private PubSubProtocol.ResponseBody responseBody;
-
-    // Constructor
-    public PubSubCallback(PubSubData pubSubData) {
-        this.pubSubData = pubSubData;
-    }
-
-    public void operationFinished(Object ctx, PubSubProtocol.ResponseBody resultOfOperation) {
-        logger.debug("PubSub call succeeded for pubSubData: {}", pubSubData);
-        // Wake up the main sync PubSub thread that is waiting for us to
-        // complete.
-        synchronized (pubSubData) {
-            this.responseBody = resultOfOperation;
-            isCallSuccessful = true;
-            pubSubData.isDone = true;
-            pubSubData.notify();
-        }
-    }
-
-    public void operationFailed(Object ctx, PubSubException exception) {
-        logger.debug("PubSub call failed with exception: {}, pubSubData: {}", exception, pubSubData);
-        // Wake up the main sync PubSub thread that is waiting for us to
-        // complete.
-        synchronized (pubSubData) {
-            isCallSuccessful = false;
-            failureException = exception;
-            pubSubData.isDone = true;
-            pubSubData.notify();
-        }
-    }
-
-    // Public getter to determine if the PubSub callback is successful or not
-    // based on the PubSub ack response from the server.
-    public boolean getIsCallSuccessful() {
-        return isCallSuccessful;
-    }
-
-    // Public getter to retrieve what the PubSubException was that occurred when
-    // the operation failed.
-    public PubSubException getFailureException() {
-        return failureException;
-    }
-
-
-    public PubSubProtocol.ResponseBody getResponseBody() {
-        return responseBody;
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java
----------------------------------------------------------------------
diff --git a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java b/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java
deleted file mode 100644
index 8b2e9d1..0000000
--- a/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java
+++ /dev/null
@@ -1,69 +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.client.handlers;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.jboss.netty.channel.Channel;
-import org.apache.hedwig.client.conf.ClientConfiguration;
-import org.apache.hedwig.client.data.PubSubData;
-import org.apache.hedwig.client.netty.HChannelManager;
-import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
-import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
-
-public class PublishResponseHandler extends AbstractResponseHandler {
-
-    private static final Logger logger = LoggerFactory.getLogger(PublishResponseHandler.class);
-
-    public PublishResponseHandler(ClientConfiguration cfg,
-                                  HChannelManager channelManager) {
-        super(cfg, channelManager);
-    }
-
-    @Override
-    public void handleResponse(PubSubResponse response, PubSubData pubSubData,
-                               Channel channel) throws Exception {
-        switch (response.getStatusCode()) {
-        case SUCCESS:
-            // Response was success so invoke the callback's operationFinished
-            // method.
-            pubSubData.operationFinishedToCallback(pubSubData.context,
-                response.hasResponseBody() ? response.getResponseBody() : null);
-            break;
-        case SERVICE_DOWN:
-            // Response was service down failure so just invoke the callback's
-            // operationFailed method.
-            pubSubData.getCallback().operationFailed(pubSubData.context, new ServiceDownException(
-                                                    "Server responded with a SERVICE_DOWN status"));
-            break;
-        case NOT_RESPONSIBLE_FOR_TOPIC:
-            // Redirect response so we'll need to repost the original Publish
-            // Request
-            handleRedirectResponse(response, pubSubData, channel);
-            break;
-        default:
-            // Consider all other status codes as errors, operation failed
-            // cases.
-            logger.error("Unexpected error response from server for PubSubResponse: " + response);
-            pubSubData.getCallback().operationFailed(pubSubData.context, new ServiceDownException(
-                                                    "Server responded with a status code of: " +
-                                                        response.getStatusCode()));
-            break;
-        }
-    }
-}


Mime
View raw message