distributedlog-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [11/30] incubator-distributedlog git commit: DL-205: Remove StatusCode dependency on DLException
Date Mon, 12 Jun 2017 15:45:18 GMT
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestRegionUnavailable.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestRegionUnavailable.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestRegionUnavailable.java
new file mode 100644
index 0000000..d0a2f88
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestRegionUnavailable.java
@@ -0,0 +1,140 @@
+/**
+ * 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.distributedlog.service;
+
+import org.apache.distributedlog.DistributedLogConfiguration;
+import org.apache.distributedlog.feature.DefaultFeatureProvider;
+import org.apache.distributedlog.service.DistributedLogCluster.DLServer;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.bookkeeper.feature.Feature;
+import org.apache.bookkeeper.feature.FeatureProvider;
+import org.apache.bookkeeper.feature.SettableFeature;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link org.apache.distributedlog.exceptions.RegionUnavailableException}.
+ */
+public class TestRegionUnavailable extends DistributedLogServerTestCase {
+
+    /**
+     * A feature provider for testing.
+     */
+    public static class TestFeatureProvider extends DefaultFeatureProvider {
+
+        public TestFeatureProvider(String rootScope,
+                                   DistributedLogConfiguration conf,
+                                   StatsLogger statsLogger) {
+            super(rootScope, conf, statsLogger);
+        }
+
+        @Override
+        protected Feature makeFeature(String featureName) {
+            if (featureName.contains(ServerFeatureKeys.REGION_STOP_ACCEPT_NEW_STREAM.name().toLowerCase())) {
+                return new SettableFeature(featureName, 10000);
+            }
+            return super.makeFeature(featureName);
+        }
+
+        @Override
+        protected FeatureProvider makeProvider(String fullScopeName) {
+            return super.makeProvider(fullScopeName);
+        }
+    }
+
+    private final int numServersPerDC = 3;
+    private final List<DLServer> localCluster;
+    private final List<DLServer> remoteCluster;
+    private TwoRegionDLClient client;
+
+    public TestRegionUnavailable() {
+        super(true);
+        this.localCluster = new ArrayList<DLServer>();
+        this.remoteCluster = new ArrayList<DLServer>();
+    }
+
+    @Before
+    @Override
+    public void setup() throws Exception {
+        DistributedLogConfiguration localConf = new DistributedLogConfiguration();
+        localConf.addConfiguration(conf);
+        localConf.setFeatureProviderClass(TestFeatureProvider.class);
+        DistributedLogConfiguration remoteConf = new DistributedLogConfiguration();
+        remoteConf.addConfiguration(conf);
+        super.setup();
+        int localPort = 9010;
+        int remotePort = 9020;
+        for (int i = 0; i < numServersPerDC; i++) {
+            localCluster.add(createDistributedLogServer(localConf, localPort + i));
+            remoteCluster.add(createDistributedLogServer(remoteConf, remotePort + i));
+        }
+        Map<SocketAddress, String> regionMap = new HashMap<SocketAddress, String>();
+        for (DLServer server : localCluster) {
+            regionMap.put(server.getAddress(), "local");
+        }
+        for (DLServer server : remoteCluster) {
+            regionMap.put(server.getAddress(), "remote");
+        }
+        client = createTwoRegionDLClient("two_regions_client", regionMap);
+
+    }
+
+    private void registerStream(String streamName) {
+        for (DLServer server : localCluster) {
+            client.localRoutingService.addHost(streamName, server.getAddress());
+        }
+        client.remoteRoutingService.addHost(streamName, remoteCluster.get(0).getAddress());
+    }
+
+    @After
+    @Override
+    public void teardown() throws Exception {
+        super.teardown();
+        if (null != client) {
+            client.shutdown();
+        }
+        for (DLServer server : localCluster) {
+            server.shutdown();
+        }
+        for (DLServer server : remoteCluster) {
+            server.shutdown();
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testRegionUnavailable() throws Exception {
+        String name = "dlserver-region-unavailable";
+        registerStream(name);
+
+        for (long i = 1; i <= 10; i++) {
+            client.dlClient.write(name, ByteBuffer.wrap(("" + i).getBytes())).get();
+        }
+
+        // check local region
+        for (DLServer server : localCluster) {
+            checkStreams(0, server);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestStatsFilter.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestStatsFilter.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestStatsFilter.java
new file mode 100644
index 0000000..c8b8bdf
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/TestStatsFilter.java
@@ -0,0 +1,58 @@
+/**
+ * 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.distributedlog.service;
+
+import static org.junit.Assert.assertEquals;
+
+import com.twitter.finagle.Service;
+import com.twitter.finagle.service.ConstantService;
+import com.twitter.util.Await;
+import com.twitter.util.Future;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link StatsFilter}.
+ */
+public class TestStatsFilter {
+
+    class RuntimeExService<Req, Rep> extends Service<Req, Rep> {
+        public Future<Rep> apply(Req request) {
+            throw new RuntimeException("test");
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testServiceSuccess() throws Exception {
+        StatsLogger stats = new NullStatsLogger();
+        StatsFilter<String, String> filter = new StatsFilter<String, String>(stats);
+        Future<String> result = filter.apply("", new ConstantService<String, String>(Future.value("result")));
+        assertEquals("result", Await.result(result));
+    }
+
+    @Test(timeout = 60000)
+    public void testServiceFailure() throws Exception {
+        StatsLogger stats = new NullStatsLogger();
+        StatsFilter<String, String> filter = new StatsFilter<String, String>(stats);
+        try {
+            filter.apply("", new RuntimeExService<String, String>());
+        } catch (RuntimeException ex) {
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestBalancerUtils.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestBalancerUtils.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestBalancerUtils.java
new file mode 100644
index 0000000..21bebb5
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestBalancerUtils.java
@@ -0,0 +1,65 @@
+/**
+ * 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.distributedlog.service.balancer;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link BalancerUtils}.
+ */
+public class TestBalancerUtils {
+
+    @Test(timeout = 60000)
+    public void testCalculateNumStreamsToRebalance() {
+        String myNode = "mynode";
+
+        // empty load distribution
+        assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance(
+                myNode, new HashMap<String, Integer>(), 0, 10));
+        // my node doesn't exist in load distribution
+        Map<String, Integer> loadDistribution = new HashMap<String, Integer>();
+        loadDistribution.put("node1", 10);
+        assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance(
+                myNode, loadDistribution, 0, 10));
+        // my node doesn't reach rebalance water mark
+        loadDistribution.clear();
+        loadDistribution.put("node1", 1);
+        loadDistribution.put(myNode, 100);
+        assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance(
+                myNode, loadDistribution, 200, 10));
+        // my node is below average in the cluster.
+        loadDistribution.clear();
+        loadDistribution.put(myNode, 1);
+        loadDistribution.put("node1", 99);
+        assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance(
+                myNode, loadDistribution, 0, 10));
+        // my node is above average in the cluster
+        assertEquals(49, BalancerUtils.calculateNumStreamsToRebalance(
+                "node1", loadDistribution, 0, 10));
+        // my node is at the tolerance range
+        loadDistribution.clear();
+        loadDistribution.put(myNode, 55);
+        loadDistribution.put("node1", 45);
+        assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance(
+                myNode, loadDistribution, 0, 10));
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestClusterBalancer.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestClusterBalancer.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestClusterBalancer.java
new file mode 100644
index 0000000..fb3fb6e
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestClusterBalancer.java
@@ -0,0 +1,189 @@
+/**
+ * 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.distributedlog.service.balancer;
+
+import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.fail;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.RateLimiter;
+import org.apache.distributedlog.client.monitor.MonitorServiceClient;
+import org.apache.distributedlog.service.DLSocketAddress;
+import org.apache.distributedlog.service.DistributedLogClient;
+import org.apache.distributedlog.service.DistributedLogCluster.DLServer;
+import org.apache.distributedlog.service.DistributedLogServerTestCase;
+import com.twitter.util.Await;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.commons.lang3.tuple.Pair;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test Case for {@link ClusterBalancer}.
+ */
+public class TestClusterBalancer extends DistributedLogServerTestCase {
+
+    private static final Logger logger = LoggerFactory.getLogger(TestClusterBalancer.class);
+
+    private final int numServers = 5;
+    private final List<DLServer> cluster;
+    private DLClient client;
+
+    public TestClusterBalancer() {
+        super(true);
+        this.cluster = new ArrayList<DLServer>();
+    }
+
+    @Before
+    @Override
+    public void setup() throws Exception {
+        super.setup();
+        int initPort = 9001;
+        for (int i = 0; i < numServers; i++) {
+            cluster.add(createDistributedLogServer(initPort + i));
+        }
+        client = createDistributedLogClient("cluster_client", Optional.<String>absent());
+    }
+
+    @After
+    @Override
+    public void teardown() throws Exception {
+        super.teardown();
+        if (null != client) {
+            client.shutdown();
+        }
+        for (DLServer server: cluster) {
+            server.shutdown();
+        }
+    }
+
+    private void initStreams(String namePrefix) {
+        logger.info("Init streams with prefix {}", namePrefix);
+        // Stream Distribution: 5, 4, 3, 2, 1
+        initStreams(namePrefix, 5, 1, 0);
+        initStreams(namePrefix, 4, 6, 1);
+        initStreams(namePrefix, 3, 10, 2);
+        initStreams(namePrefix, 2, 13, 3);
+        initStreams(namePrefix, 1, 15, 4);
+    }
+
+    private void initStreams(String namePrefix, int numStreams, int streamId, int proxyId) {
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + (streamId++);
+            client.routingService.addHost(name, cluster.get(proxyId).getAddress());
+        }
+    }
+
+    private void writeStreams(String namePrefix) throws Exception {
+        logger.info("Write streams with prefix {}", namePrefix);
+        writeStreams(namePrefix, 5, 1);
+        writeStreams(namePrefix, 4, 6);
+        writeStreams(namePrefix, 3, 10);
+        writeStreams(namePrefix, 2, 13);
+        writeStreams(namePrefix, 1, 15);
+    }
+
+    private void writeStreams(String namePrefix, int numStreams, int streamId) throws Exception {
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + (streamId++);
+            try {
+                Await.result(client.dlClient.write(name, ByteBuffer.wrap(name.getBytes(UTF_8))));
+            } catch (Exception e) {
+                logger.error("Error writing stream {} : ", name, e);
+                throw e;
+            }
+        }
+    }
+
+    private void validateStreams(String namePrefix) throws Exception {
+        logger.info("Validate streams with prefix {}", namePrefix);
+        validateStreams(namePrefix, 5, 1, 0);
+        validateStreams(namePrefix, 4, 6, 1);
+        validateStreams(namePrefix, 3, 10, 2);
+        validateStreams(namePrefix, 2, 13, 3);
+        validateStreams(namePrefix, 1, 15, 4);
+    }
+
+    private void validateStreams(String namePrefix, int numStreams, int streamId, int proxyIdx) {
+        Set<String> expectedStreams = new HashSet<String>();
+        for (int i = 0; i < numStreams; i++) {
+            expectedStreams.add(namePrefix + (streamId++));
+        }
+        checkStreams(expectedStreams, cluster.get(proxyIdx));
+    }
+
+    @Ignore
+    @Test(timeout = 60000)
+    public void testBalanceAll() throws Exception {
+        String namePrefix = "clusterbalancer-balance-all-";
+
+        initStreams(namePrefix);
+        writeStreams(namePrefix);
+        validateStreams(namePrefix);
+
+        Optional<RateLimiter> rateLimiter = Optional.absent();
+
+        Balancer balancer = new ClusterBalancer(client.dlClientBuilder,
+                Pair.of((DistributedLogClient) client.dlClient, (MonitorServiceClient) client.dlClient));
+        logger.info("Rebalancing from 'unknown' target");
+        try {
+            balancer.balanceAll("unknown", 10, rateLimiter);
+            fail("Should fail on balanceAll from 'unknown' target.");
+        } catch (IllegalArgumentException iae) {
+            // expected
+        }
+        validateStreams(namePrefix);
+
+        logger.info("Rebalancing from 'unexisted' host");
+        String addr = DLSocketAddress.toString(DLSocketAddress.getSocketAddress(9999));
+        balancer.balanceAll(addr, 10, rateLimiter);
+        validateStreams(namePrefix);
+
+        addr = DLSocketAddress.toString(cluster.get(0).getAddress());
+        logger.info("Rebalancing from host {}.", addr);
+        balancer.balanceAll(addr, 10, rateLimiter);
+        checkStreams(0, cluster.get(0));
+        checkStreams(4, cluster.get(1));
+        checkStreams(3, cluster.get(2));
+        checkStreams(4, cluster.get(3));
+        checkStreams(4, cluster.get(4));
+
+        addr = DLSocketAddress.toString(cluster.get(2).getAddress());
+        logger.info("Rebalancing from host {}.", addr);
+        balancer.balanceAll(addr, 10, rateLimiter);
+        checkStreams(3, cluster.get(0));
+        checkStreams(4, cluster.get(1));
+        checkStreams(0, cluster.get(2));
+        checkStreams(4, cluster.get(3));
+        checkStreams(4, cluster.get(4));
+
+        logger.info("Rebalancing the cluster");
+        balancer.balance(0, 0.0f, 10, rateLimiter);
+        for (int i = 0; i < 5; i++) {
+            checkStreams(3, cluster.get(i));
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestCountBasedStreamChooser.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestCountBasedStreamChooser.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestCountBasedStreamChooser.java
new file mode 100644
index 0000000..6734083
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestCountBasedStreamChooser.java
@@ -0,0 +1,204 @@
+/**
+ * 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.distributedlog.service.balancer;
+
+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 static org.junit.Assert.fail;
+
+import com.google.common.collect.Sets;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link CountBasedStreamChooser}.
+ */
+public class TestCountBasedStreamChooser {
+
+    @Test(timeout = 60000)
+    public void testEmptyStreamDistribution() {
+        try {
+            new CountBasedStreamChooser(new HashMap<SocketAddress, Set<String>>());
+            fail("Should fail constructing stream chooser if the stream distribution is empty");
+        } catch (IllegalArgumentException iae) {
+            // expected
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testMultipleHostsWithEmptyStreams() {
+        for (int i = 1; i <= 3; i++) {
+            Map<SocketAddress, Set<String>> streamDistribution = new HashMap<SocketAddress, Set<String>>();
+            int port = 1000;
+            for (int j = 0; j < i; j++) {
+                SocketAddress address = new InetSocketAddress("127.0.0.1", port + j);
+                streamDistribution.put(address, new HashSet<String>());
+            }
+
+            CountBasedStreamChooser chooser = new CountBasedStreamChooser(streamDistribution);
+            for (int k = 0; k < i + 1; k++) {
+                assertNull(chooser.choose());
+            }
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testSingleHostWithStreams() {
+        for (int i = 0; i < 3; i++) {
+            Map<SocketAddress, Set<String>> streamDistribution = new HashMap<SocketAddress, Set<String>>();
+
+            Set<String> streams = new HashSet<String>();
+            for (int j = 0; j < 3; j++) {
+                streams.add("SingleHostStream-" + j);
+            }
+
+            int port = 1000;
+            SocketAddress address = new InetSocketAddress("127.0.0.1", port);
+            streamDistribution.put(address, streams);
+
+            for (int k = 1; k <= i; k++) {
+                address = new InetSocketAddress("127.0.0.1", port + k);
+                streamDistribution.put(address, new HashSet<String>());
+            }
+
+            Set<String> choosenStreams = new HashSet<String>();
+
+            CountBasedStreamChooser chooser = new CountBasedStreamChooser(streamDistribution);
+            for (int l = 0; l < 3 + i + 1; l++) {
+                String s = chooser.choose();
+                if (null != s) {
+                    choosenStreams.add(s);
+                }
+            }
+
+            assertEquals(streams.size(), choosenStreams.size());
+            assertTrue(Sets.difference(streams, choosenStreams).immutableCopy().isEmpty());
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testHostsHaveSameNumberStreams() {
+        Map<SocketAddress, Set<String>> streamDistribution = new HashMap<SocketAddress, Set<String>>();
+        Set<String> allStreams = new HashSet<String>();
+
+        int numHosts = 3;
+        int numStreamsPerHost = 3;
+
+        int port = 1000;
+        for (int i = 1; i <= numHosts; i++) {
+            SocketAddress address = new InetSocketAddress("127.0.0.1", port + i);
+            Set<String> streams = new HashSet<String>();
+
+            for (int j = 1; j <= numStreamsPerHost; j++) {
+                String streamName = "HostsHaveSameNumberStreams-" + i + "-" + j;
+                streams.add(streamName);
+                allStreams.add(streamName);
+            }
+
+            streamDistribution.put(address, streams);
+        }
+
+        Set<String> streamsChoosen = new HashSet<String>();
+        CountBasedStreamChooser chooser = new CountBasedStreamChooser(streamDistribution);
+        for (int i = 1; i <= numStreamsPerHost; i++) {
+            for (int j = 1; j <= numHosts; j++) {
+                String s = chooser.choose();
+                assertNotNull(s);
+                streamsChoosen.add(s);
+            }
+            for (int j = 0; j < numHosts; j++) {
+                assertEquals(numStreamsPerHost - i, chooser.streamsDistribution.get(j).getRight().size());
+            }
+        }
+        assertNull(chooser.choose());
+        assertEquals(numHosts * numStreamsPerHost, streamsChoosen.size());
+        assertTrue(Sets.difference(allStreams, streamsChoosen).isEmpty());
+    }
+
+    @Test(timeout = 60000)
+    public void testHostsHaveDifferentNumberStreams() {
+        Map<SocketAddress, Set<String>> streamDistribution = new HashMap<SocketAddress, Set<String>>();
+        Set<String> allStreams = new HashSet<String>();
+
+        int numHosts = 6;
+        int maxStreamsPerHost = 4;
+
+        int port = 1000;
+        for (int i = 0; i < numHosts; i++) {
+            int group = i / 2;
+            int numStreamsThisGroup = maxStreamsPerHost - group;
+
+            SocketAddress address = new InetSocketAddress("127.0.0.1", port + i);
+            Set<String> streams = new HashSet<String>();
+
+            for (int j = 1; j <= numStreamsThisGroup; j++) {
+                String streamName = "HostsHaveDifferentNumberStreams-" + i + "-" + j;
+                streams.add(streamName);
+                allStreams.add(streamName);
+            }
+
+            streamDistribution.put(address, streams);
+        }
+
+        Set<String> streamsChoosen = new HashSet<String>();
+        CountBasedStreamChooser chooser = new CountBasedStreamChooser(streamDistribution);
+
+        for (int i = 0; i < allStreams.size(); i++) {
+            String s = chooser.choose();
+            assertNotNull(s);
+            streamsChoosen.add(s);
+        }
+        assertNull(chooser.choose());
+        assertEquals(allStreams.size(), streamsChoosen.size());
+        assertTrue(Sets.difference(allStreams, streamsChoosen).isEmpty());
+    }
+
+    @Test(timeout = 60000)
+    public void testLimitedStreamChooser() {
+        Map<SocketAddress, Set<String>> streamDistribution = new HashMap<SocketAddress, Set<String>>();
+
+        Set<String> streams = new HashSet<String>();
+        for (int j = 0; j < 10; j++) {
+            streams.add("SingleHostStream-" + j);
+        }
+
+        int port = 1000;
+        SocketAddress address = new InetSocketAddress("127.0.0.1", port);
+        streamDistribution.put(address, streams);
+
+        Set<String> choosenStreams = new HashSet<String>();
+
+        CountBasedStreamChooser underlying = new CountBasedStreamChooser(streamDistribution);
+        LimitedStreamChooser chooser = LimitedStreamChooser.of(underlying, 1);
+        for (int l = 0; l < 10; l++) {
+            String s = chooser.choose();
+            if (null != s) {
+                choosenStreams.add(s);
+            }
+        }
+
+        assertEquals(1, choosenStreams.size());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestSimpleBalancer.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestSimpleBalancer.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestSimpleBalancer.java
new file mode 100644
index 0000000..73fa98a
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestSimpleBalancer.java
@@ -0,0 +1,180 @@
+/**
+ * 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.distributedlog.service.balancer;
+
+import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.RateLimiter;
+import org.apache.distributedlog.service.DistributedLogCluster.DLServer;
+import org.apache.distributedlog.service.DistributedLogServerTestCase;
+import com.twitter.util.Await;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test Case for {@link SimpleBalancer}.
+ */
+public class TestSimpleBalancer extends DistributedLogServerTestCase {
+
+    private static final Logger logger = LoggerFactory.getLogger(TestSimpleBalancer.class);
+
+    DLClient targetClient;
+    DLServer targetServer;
+
+    public TestSimpleBalancer() {
+        super(true);
+    }
+
+    @Before
+    @Override
+    public void setup() throws Exception {
+        super.setup();
+        targetServer = createDistributedLogServer(7003);
+        targetClient = createDistributedLogClient("target", Optional.<String>absent());
+    }
+
+    @After
+    @Override
+    public void teardown() throws Exception {
+        super.teardown();
+        if (null != targetClient) {
+            targetClient.shutdown();
+        }
+        if (null != targetServer) {
+            targetServer.shutdown();
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testBalanceAll() throws Exception {
+        String namePrefix = "simplebalancer-balance-all-";
+        int numStreams = 10;
+
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            // src client
+            dlClient.routingService.addHost(name, dlServer.getAddress());
+            // target client
+            targetClient.routingService.addHost(name, targetServer.getAddress());
+        }
+
+        // write to multiple streams
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            Await.result(dlClient.dlClient.write(name, ByteBuffer.wrap(("" + i).getBytes(UTF_8))));
+        }
+
+        // validation
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            checkStream(name, dlClient, dlServer, 1, numStreams, numStreams, true, true);
+            checkStream(name, targetClient, targetServer, 0, 0, 0, false, false);
+        }
+
+        Optional<RateLimiter> rateLimiter = Optional.absent();
+
+        Balancer balancer = new SimpleBalancer("source", dlClient.dlClient, dlClient.dlClient,
+                                               "target", targetClient.dlClient, targetClient.dlClient);
+        logger.info("Rebalancing from 'unknown' target");
+        try {
+            balancer.balanceAll("unknown", 10, rateLimiter);
+            fail("Should fail on balanceAll from 'unknown' target.");
+        } catch (IllegalArgumentException iae) {
+            // expected
+        }
+
+        // nothing to balance from 'target'
+        logger.info("Rebalancing from 'target' target");
+        balancer.balanceAll("target", 1, rateLimiter);
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            checkStream(name, dlClient, dlServer, 1, numStreams, numStreams, true, true);
+            checkStream(name, targetClient, targetServer, 0, 0, 0, false, false);
+        }
+
+        // balance all streams from 'source'
+        logger.info("Rebalancing from 'source' target");
+        balancer.balanceAll("source", 10, rateLimiter);
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            checkStream(name, targetClient, targetServer, 1, numStreams, numStreams, true, true);
+            checkStream(name, dlClient, dlServer, 0, 0, 0, false, false);
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testBalanceStreams() throws Exception {
+        String namePrefix = "simplebalancer-balance-streams-";
+        int numStreams = 10;
+
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            // src client
+            dlClient.routingService.addHost(name, dlServer.getAddress());
+            // target client
+            targetClient.routingService.addHost(name, targetServer.getAddress());
+        }
+
+        // write to multiple streams
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            Await.result(dlClient.dlClient.write(name, ByteBuffer.wrap(("" + i).getBytes(UTF_8))));
+        }
+
+        // validation
+        for (int i = 0; i < numStreams; i++) {
+            String name = namePrefix + i;
+            checkStream(name, dlClient, dlServer, 1, numStreams, numStreams, true, true);
+            checkStream(name, targetClient, targetServer, 0, 0, 0, false, false);
+        }
+
+        Optional<RateLimiter> rateLimiter = Optional.absent();
+
+        Balancer balancer = new SimpleBalancer("source", dlClient.dlClient, dlClient.dlClient,
+                                               "target", targetClient.dlClient, targetClient.dlClient);
+
+        // balance all streams from 'source'
+        logger.info("Rebalancing streams between targets");
+        balancer.balance(0, 0, 10, rateLimiter);
+
+        Set<String> sourceStreams = getAllStreamsFromDistribution(getStreamOwnershipDistribution(dlClient));
+        Set<String> targetStreams = getAllStreamsFromDistribution(getStreamOwnershipDistribution(targetClient));
+
+        assertEquals(numStreams / 2, sourceStreams.size());
+        assertEquals(numStreams / 2, targetStreams.size());
+
+        for (String name : sourceStreams) {
+            checkStream(name, dlClient, dlServer, 1, numStreams / 2, numStreams / 2, true, true);
+            checkStream(name, targetClient, targetServer, 1, numStreams / 2, numStreams / 2, false, false);
+        }
+
+        for (String name : targetStreams) {
+            checkStream(name, targetClient, targetServer, 1, numStreams / 2, numStreams / 2, true, true);
+            checkStream(name, dlClient, dlServer, 1, numStreams / 2, numStreams / 2, false, false);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestStreamMover.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestStreamMover.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestStreamMover.java
new file mode 100644
index 0000000..ce7b2c1
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/balancer/TestStreamMover.java
@@ -0,0 +1,86 @@
+/**
+ * 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.distributedlog.service.balancer;
+
+import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertTrue;
+
+import com.google.common.base.Optional;
+import org.apache.distributedlog.service.DistributedLogClient;
+import org.apache.distributedlog.service.DistributedLogCluster.DLServer;
+import org.apache.distributedlog.service.DistributedLogServerTestCase;
+import com.twitter.util.Await;
+import java.nio.ByteBuffer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link StreamMover}.
+ */
+public class TestStreamMover extends DistributedLogServerTestCase {
+
+    DLClient targetClient;
+    DLServer targetServer;
+
+    public TestStreamMover() {
+        super(true);
+    }
+
+    @Before
+    @Override
+    public void setup() throws Exception {
+        super.setup();
+        targetServer = createDistributedLogServer(7003);
+        targetClient = createDistributedLogClient("target", Optional.<String>absent());
+    }
+
+    @After
+    @Override
+    public void teardown() throws Exception {
+        super.teardown();
+        if (null != targetClient) {
+            targetClient.shutdown();
+        }
+        if (null != targetServer) {
+            targetServer.shutdown();
+        }
+    }
+
+    @Test(timeout = 60000)
+    public void testMoveStream() throws Exception {
+        String name = "dlserver-move-stream";
+
+        // src client
+        dlClient.routingService.addHost(name, dlServer.getAddress());
+        // target client
+        targetClient.routingService.addHost(name, targetServer.getAddress());
+
+        // src client write a record to that stream
+        Await.result(((DistributedLogClient) dlClient.dlClient).write(name, ByteBuffer.wrap("1".getBytes(UTF_8))));
+        checkStream(name, dlClient, dlServer, 1, 1, 1, true, true);
+        checkStream(name, targetClient, targetServer, 0, 0, 0, false, false);
+
+        StreamMover streamMover = new StreamMoverImpl("source", dlClient.dlClient, dlClient.dlClient,
+                                                      "target", targetClient.dlClient, targetClient.dlClient);
+        assertTrue(streamMover.moveStream(name));
+        checkStream(name, dlClient, dlServer, 0, 0, 0, false, false);
+        checkStream(name, targetClient, targetServer, 1, 1, 1, true, true);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestServerConfiguration.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestServerConfiguration.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestServerConfiguration.java
new file mode 100644
index 0000000..71dfa45
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestServerConfiguration.java
@@ -0,0 +1,68 @@
+/**
+ * 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.distributedlog.service.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+/**
+ * Test Case for {@link ServerConfiguration}.
+ */
+public class TestServerConfiguration {
+
+    @Test(timeout = 60000, expected = IllegalArgumentException.class)
+    public void testUnassignedShardId() {
+        new ServerConfiguration().validate();
+    }
+
+    @Test(timeout = 60000)
+    public void testAssignedShardId() {
+        ServerConfiguration conf = new ServerConfiguration();
+        conf.setServerShardId(100);
+        conf.validate();
+        assertEquals(100, conf.getServerShardId());
+    }
+
+    @Test(timeout = 60000, expected = IllegalArgumentException.class)
+    public void testInvalidServerThreads() {
+        ServerConfiguration conf = new ServerConfiguration();
+        conf.setServerShardId(100);
+        conf.setServerThreads(-1);
+        conf.validate();
+    }
+
+    @Test(timeout = 60000, expected = IllegalArgumentException.class)
+    public void testInvalidDlsnVersion() {
+        ServerConfiguration conf = new ServerConfiguration();
+        conf.setServerShardId(100);
+        conf.setDlsnVersion((byte) 9999);
+        conf.validate();
+    }
+
+    @Test(timeout = 60000)
+    public void testUseHostnameAsAllocatorPoolName() {
+        ServerConfiguration conf = new ServerConfiguration();
+        assertFalse("Should not use hostname by default", conf.isUseHostnameAsAllocatorPoolName());
+        conf.setUseHostnameAsAllocatorPoolName(true);
+        assertTrue("Should use hostname now", conf.isUseHostnameAsAllocatorPoolName());
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestStreamConfigProvider.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestStreamConfigProvider.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestStreamConfigProvider.java
new file mode 100644
index 0000000..bdbde11
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/config/TestStreamConfigProvider.java
@@ -0,0 +1,140 @@
+/**
+ * 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.distributedlog.service.config;
+
+import static org.apache.distributedlog.DistributedLogConfiguration.BKDL_RETENTION_PERIOD_IN_HOURS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.distributedlog.config.DynamicDistributedLogConfiguration;
+import org.apache.distributedlog.config.PropertiesWriter;
+import org.apache.distributedlog.service.streamset.IdentityStreamPartitionConverter;
+import org.apache.distributedlog.service.streamset.StreamPartitionConverter;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link StreamConfigProvider}.
+ */
+public class TestStreamConfigProvider {
+    private static final String DEFAULT_CONFIG_DIR = "conf";
+    private final String defaultConfigPath;
+    private final ScheduledExecutorService configExecutorService;
+
+    public TestStreamConfigProvider() throws Exception {
+        this.configExecutorService = Executors.newScheduledThreadPool(1,
+                new ThreadFactoryBuilder().setNameFormat("DistributedLogService-Dyncfg-%d").build());
+        PropertiesWriter writer = new PropertiesWriter();
+        writer.save();
+        this.defaultConfigPath = writer.getFile().getPath();
+    }
+
+    StreamConfigProvider getServiceProvider(StreamPartitionConverter converter)
+            throws Exception {
+        return getServiceProvider(converter, DEFAULT_CONFIG_DIR);
+    }
+
+    StreamConfigProvider getServiceProvider(
+            StreamPartitionConverter converter,
+            String configPath,
+            String defaultPath) throws Exception {
+        return new ServiceStreamConfigProvider(
+                configPath,
+                defaultPath,
+                converter,
+                configExecutorService,
+                1,
+                TimeUnit.SECONDS);
+    }
+
+    StreamConfigProvider getServiceProvider(
+            StreamPartitionConverter converter,
+            String configPath) throws Exception {
+        return getServiceProvider(converter, configPath, defaultConfigPath);
+    }
+
+    StreamConfigProvider getDefaultProvider(String configFile) throws Exception {
+        return new DefaultStreamConfigProvider(configFile, configExecutorService, 1, TimeUnit.SECONDS);
+    }
+
+    StreamConfigProvider getNullProvider() throws Exception {
+        return new NullStreamConfigProvider();
+    }
+
+    @Test(timeout = 60000)
+    public void testServiceProviderWithConfigRouters() throws Exception {
+        getServiceProvider(new IdentityStreamPartitionConverter());
+    }
+
+    @Test(timeout = 60000)
+    public void testServiceProviderWithMissingConfig() throws Exception {
+        StreamConfigProvider provider = getServiceProvider(new IdentityStreamPartitionConverter());
+        Optional<DynamicDistributedLogConfiguration> config = provider.getDynamicStreamConfig("stream1");
+        assertTrue(config.isPresent());
+    }
+
+    @Test(timeout = 60000)
+    public void testServiceProviderWithDefaultConfigPath() throws Exception {
+        // Default config with property set.
+        PropertiesWriter writer1 = new PropertiesWriter();
+        writer1.setProperty("rpsStreamAcquireServiceLimit", "191919");
+        writer1.save();
+        String fallbackConfPath1 = writer1.getFile().getPath();
+        StreamConfigProvider provider1 = getServiceProvider(new IdentityStreamPartitionConverter(),
+                DEFAULT_CONFIG_DIR, fallbackConfPath1);
+        Optional<DynamicDistributedLogConfiguration> config1 = provider1.getDynamicStreamConfig("stream1");
+
+        // Empty default config.
+        PropertiesWriter writer2 = new PropertiesWriter();
+        writer2.save();
+        String fallbackConfPath2 = writer2.getFile().getPath();
+        StreamConfigProvider provider2 = getServiceProvider(new IdentityStreamPartitionConverter(),
+                DEFAULT_CONFIG_DIR, fallbackConfPath2);
+        Optional<DynamicDistributedLogConfiguration> config2 = provider2.getDynamicStreamConfig("stream1");
+
+        assertEquals(191919, config1.get().getRpsStreamAcquireServiceLimit());
+        assertEquals(-1, config2.get().getRpsStreamAcquireServiceLimit());
+    }
+
+    @Test(timeout = 60000)
+    public void testDefaultProvider() throws Exception {
+        PropertiesWriter writer = new PropertiesWriter();
+        writer.setProperty(BKDL_RETENTION_PERIOD_IN_HOURS, "99");
+        writer.save();
+        StreamConfigProvider provider = getDefaultProvider(writer.getFile().getPath());
+        Optional<DynamicDistributedLogConfiguration> config1 = provider.getDynamicStreamConfig("stream1");
+        Optional<DynamicDistributedLogConfiguration> config2 = provider.getDynamicStreamConfig("stream2");
+        assertTrue(config1.isPresent());
+        assertTrue(config1.get() == config2.get());
+        assertEquals(99, config1.get().getRetentionPeriodHours());
+    }
+
+    @Test(timeout = 60000)
+    public void testNullProvider() throws Exception {
+        StreamConfigProvider provider = getNullProvider();
+        Optional<DynamicDistributedLogConfiguration> config1 = provider.getDynamicStreamConfig("stream1");
+        Optional<DynamicDistributedLogConfiguration> config2 = provider.getDynamicStreamConfig("stream2");
+        assertFalse(config1.isPresent());
+        assertTrue(config1 == config2);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestLeastLoadPlacementPolicy.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestLeastLoadPlacementPolicy.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestLeastLoadPlacementPolicy.java
new file mode 100644
index 0000000..5f5ecd4
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestLeastLoadPlacementPolicy.java
@@ -0,0 +1,176 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.distributedlog.service.placement;
+
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.distributedlog.client.routing.RoutingService;
+import org.apache.distributedlog.namespace.DistributedLogNamespace;
+import com.twitter.util.Await;
+import com.twitter.util.Duration;
+import com.twitter.util.Future;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.LinkedHashSet;
+import java.util.Random;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/**
+ * Test Case for {@link LeastLoadPlacementPolicy}.
+ */
+public class TestLeastLoadPlacementPolicy {
+
+    @Test(timeout = 10000)
+    public void testCalculateBalances() throws Exception {
+        int numSevers = new Random().nextInt(20) + 1;
+        int numStreams = new Random().nextInt(200) + 1;
+        RoutingService mockRoutingService = mock(RoutingService.class);
+        DistributedLogNamespace mockNamespace = mock(DistributedLogNamespace.class);
+        LeastLoadPlacementPolicy leastLoadPlacementPolicy = new LeastLoadPlacementPolicy(
+            new EqualLoadAppraiser(),
+            mockRoutingService,
+            mockNamespace,
+            null,
+            Duration.fromSeconds(600),
+            new NullStatsLogger());
+        TreeSet<ServerLoad> serverLoads =
+            Await.result(leastLoadPlacementPolicy.calculate(generateServers(numSevers), generateStreams(numStreams)));
+        long lowLoadPerServer = numStreams / numSevers;
+        long highLoadPerServer = lowLoadPerServer + 1;
+        for (ServerLoad serverLoad : serverLoads) {
+            long load = serverLoad.getLoad();
+            assertEquals(load, serverLoad.getStreamLoads().size());
+            assertTrue(String.format("Load %d is not between %d and %d",
+                load, lowLoadPerServer, highLoadPerServer), load == lowLoadPerServer || load == highLoadPerServer);
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testRefreshAndPlaceStream() throws Exception {
+        int numSevers = new Random().nextInt(20) + 1;
+        int numStreams = new Random().nextInt(200) + 1;
+        RoutingService mockRoutingService = mock(RoutingService.class);
+        when(mockRoutingService.getHosts()).thenReturn(generateSocketAddresses(numSevers));
+        DistributedLogNamespace mockNamespace = mock(DistributedLogNamespace.class);
+        try {
+            when(mockNamespace.getLogs()).thenReturn(generateStreams(numStreams).iterator());
+        } catch (IOException e) {
+            fail();
+        }
+        PlacementStateManager mockPlacementStateManager = mock(PlacementStateManager.class);
+        LeastLoadPlacementPolicy leastLoadPlacementPolicy = new LeastLoadPlacementPolicy(
+            new EqualLoadAppraiser(),
+            mockRoutingService,
+            mockNamespace,
+            mockPlacementStateManager,
+            Duration.fromSeconds(600),
+            new NullStatsLogger());
+        leastLoadPlacementPolicy.refresh();
+
+        final ArgumentCaptor<TreeSet> captor = ArgumentCaptor.forClass(TreeSet.class);
+        verify(mockPlacementStateManager).saveOwnership(captor.capture());
+        TreeSet<ServerLoad> serverLoads = (TreeSet<ServerLoad>) captor.getValue();
+        ServerLoad next = serverLoads.first();
+        String serverPlacement = Await.result(leastLoadPlacementPolicy.placeStream("newstream1"));
+        assertEquals(next.getServer(), serverPlacement);
+    }
+
+    @Test(timeout = 10000)
+    public void testCalculateUnequalWeight() throws Exception {
+        int numSevers = new Random().nextInt(20) + 1;
+        int numStreams = new Random().nextInt(200) + 1;
+    /* use AtomicInteger to have a final object in answer method */
+        final AtomicInteger maxLoad = new AtomicInteger(Integer.MIN_VALUE);
+        RoutingService mockRoutingService = mock(RoutingService.class);
+        DistributedLogNamespace mockNamespace = mock(DistributedLogNamespace.class);
+        LoadAppraiser mockLoadAppraiser = mock(LoadAppraiser.class);
+        when(mockLoadAppraiser.getStreamLoad(anyString())).then(new Answer<Future<StreamLoad>>() {
+            @Override
+            public Future<StreamLoad> answer(InvocationOnMock invocationOnMock) throws Throwable {
+                int load = new Random().nextInt(100000);
+                if (load > maxLoad.get()) {
+                    maxLoad.set(load);
+                }
+                return Future.value(new StreamLoad(invocationOnMock.getArguments()[0].toString(), load));
+            }
+        });
+        LeastLoadPlacementPolicy leastLoadPlacementPolicy = new LeastLoadPlacementPolicy(
+            mockLoadAppraiser,
+            mockRoutingService,
+            mockNamespace,
+            null,
+            Duration.fromSeconds(600),
+            new NullStatsLogger());
+        TreeSet<ServerLoad> serverLoads =
+            Await.result(leastLoadPlacementPolicy.calculate(generateServers(numSevers), generateStreams(numStreams)));
+        long highestLoadSeen = Long.MIN_VALUE;
+        long lowestLoadSeen = Long.MAX_VALUE;
+        for (ServerLoad serverLoad : serverLoads) {
+            long load = serverLoad.getLoad();
+            if (load < lowestLoadSeen) {
+                lowestLoadSeen = load;
+            }
+            if (load > highestLoadSeen) {
+                highestLoadSeen = load;
+            }
+        }
+        assertTrue("Unexpected placement for " + numStreams + " streams to "
+                + numSevers + " servers : highest load = " + highestLoadSeen
+                + ", lowest load = " + lowestLoadSeen + ", max stream load = " + maxLoad.get(),
+            highestLoadSeen - lowestLoadSeen <= maxLoad.get());
+    }
+
+    private Set<SocketAddress> generateSocketAddresses(int num) {
+        LinkedHashSet<SocketAddress> socketAddresses = new LinkedHashSet<SocketAddress>();
+        for (int i = 0; i < num; i++) {
+            socketAddresses.add(new InetSocketAddress(i));
+        }
+        return socketAddresses;
+    }
+
+    private Set<String> generateStreams(int num) {
+        LinkedHashSet<String> streams = new LinkedHashSet<String>();
+        for (int i = 0; i < num; i++) {
+            streams.add("stream_" + i);
+        }
+        return streams;
+    }
+
+    private Set<String> generateServers(int num) {
+        LinkedHashSet<String> servers = new LinkedHashSet<String>();
+        for (int i = 0; i < num; i++) {
+            servers.add("server_" + i);
+        }
+        return servers;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestServerLoad.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestServerLoad.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestServerLoad.java
new file mode 100644
index 0000000..5bd234f
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestServerLoad.java
@@ -0,0 +1,50 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.distributedlog.service.placement;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link ServerLoad}.
+ */
+public class TestServerLoad {
+
+    @Test(timeout = 60000)
+    public void testSerializeDeserialize() throws IOException {
+        final ServerLoad serverLoad = new ServerLoad("th1s1s@s3rv3rn@m3");
+        for (int i = 0; i < 20; i++) {
+            serverLoad.addStream(new StreamLoad("stream-" + i, i));
+        }
+        assertEquals(serverLoad, ServerLoad.deserialize(serverLoad.serialize()));
+    }
+
+    @Test(timeout = 60000)
+    public void testGetLoad() throws IOException {
+        final ServerLoad serverLoad = new ServerLoad("th1s1s@s3rv3rn@m3");
+        assertEquals(0, serverLoad.getLoad());
+        serverLoad.addStream(new StreamLoad("stream-" + 1, 3));
+        assertEquals(3, serverLoad.getLoad());
+        serverLoad.addStream(new StreamLoad("stream-" + 2, 7));
+        assertEquals(10, serverLoad.getLoad());
+        serverLoad.addStream(new StreamLoad("stream-" + 3, 1));
+        assertEquals(11, serverLoad.getLoad());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestStreamLoad.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestStreamLoad.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestStreamLoad.java
new file mode 100644
index 0000000..36a6fed
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestStreamLoad.java
@@ -0,0 +1,37 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.distributedlog.service.placement;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link StreamLoad}.
+ */
+public class TestStreamLoad {
+
+    @Test(timeout = 10000)
+    public void testSerializeDeserialize() throws IOException {
+        final String streamName = "aHellaRandomStreamName";
+        final int load = 1337;
+        final StreamLoad streamLoad = new StreamLoad(streamName, load);
+        assertEquals(streamLoad, StreamLoad.deserialize(streamLoad.serialize()));
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestZKPlacementStateManager.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestZKPlacementStateManager.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestZKPlacementStateManager.java
new file mode 100644
index 0000000..07ec5a5
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/placement/TestZKPlacementStateManager.java
@@ -0,0 +1,136 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.distributedlog.service.placement;
+
+import static org.apache.distributedlog.LocalDLMEmulator.DLOG_NAMESPACE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.apache.distributedlog.DistributedLogConfiguration;
+import java.io.IOException;
+import java.net.URI;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.curator.test.TestingServer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test Case for {@link ZKPlacementStateManager}.
+ */
+public class TestZKPlacementStateManager {
+    private TestingServer zkTestServer;
+    private String zkServers;
+    private URI uri;
+    private ZKPlacementStateManager zkPlacementStateManager;
+
+    @Before
+    public void startZookeeper() throws Exception {
+        zkTestServer = new TestingServer(2181);
+        zkServers = "127.0.0.1:2181";
+        uri = new URI("distributedlog-bk://" + zkServers + DLOG_NAMESPACE + "/bknamespace");
+        zkPlacementStateManager =
+            new ZKPlacementStateManager(uri, new DistributedLogConfiguration(), NullStatsLogger.INSTANCE);
+    }
+
+    @Test(timeout = 60000)
+    public void testSaveLoad() throws Exception {
+        TreeSet<ServerLoad> ownerships = new TreeSet<ServerLoad>();
+        zkPlacementStateManager.saveOwnership(ownerships);
+        SortedSet<ServerLoad> loadedOwnerships = zkPlacementStateManager.loadOwnership();
+        assertEquals(ownerships, loadedOwnerships);
+
+        ownerships.add(new ServerLoad("emptyServer"));
+        zkPlacementStateManager.saveOwnership(ownerships);
+        loadedOwnerships = zkPlacementStateManager.loadOwnership();
+        assertEquals(ownerships, loadedOwnerships);
+
+        ServerLoad sl1 = new ServerLoad("server1");
+        sl1.addStream(new StreamLoad("stream1", 3));
+        sl1.addStream(new StreamLoad("stream2", 4));
+        ServerLoad sl2 = new ServerLoad("server2");
+        sl2.addStream(new StreamLoad("stream3", 1));
+        ownerships.add(sl1);
+        ownerships.add(sl2);
+        zkPlacementStateManager.saveOwnership(ownerships);
+        loadedOwnerships = zkPlacementStateManager.loadOwnership();
+        assertEquals(ownerships, loadedOwnerships);
+
+        loadedOwnerships.remove(sl1);
+        zkPlacementStateManager.saveOwnership(ownerships);
+        loadedOwnerships = zkPlacementStateManager.loadOwnership();
+        assertEquals(ownerships, loadedOwnerships);
+    }
+
+    private TreeSet<ServerLoad> waitForServerLoadsNotificationAsc(
+        LinkedBlockingQueue<TreeSet<ServerLoad>> notificationQueue,
+        int expectedNumServerLoads) throws InterruptedException {
+        TreeSet<ServerLoad> notification = notificationQueue.take();
+        assertNotNull(notification);
+        while (notification.size() < expectedNumServerLoads) {
+            notification = notificationQueue.take();
+        }
+        assertEquals(expectedNumServerLoads, notification.size());
+        return notification;
+    }
+
+    @Test(timeout = 60000)
+    public void testWatchIndefinitely() throws Exception {
+        TreeSet<ServerLoad> ownerships = new TreeSet<ServerLoad>();
+        ownerships.add(new ServerLoad("server1"));
+        final LinkedBlockingQueue<TreeSet<ServerLoad>> serverLoadNotifications =
+            new LinkedBlockingQueue<TreeSet<ServerLoad>>();
+        PlacementStateManager.PlacementCallback callback = new PlacementStateManager.PlacementCallback() {
+            @Override
+            public void callback(TreeSet<ServerLoad> serverLoads) {
+                serverLoadNotifications.add(serverLoads);
+            }
+        };
+        zkPlacementStateManager.saveOwnership(ownerships); // need to initialize the zk path before watching
+        zkPlacementStateManager.watch(callback);
+        // cannot verify the callback here as it may call before the verify is called
+
+        zkPlacementStateManager.saveOwnership(ownerships);
+        assertEquals(ownerships, waitForServerLoadsNotificationAsc(serverLoadNotifications, 1));
+
+        ServerLoad server2 = new ServerLoad("server2");
+        server2.addStream(new StreamLoad("hella-important-stream", 415));
+        ownerships.add(server2);
+        zkPlacementStateManager.saveOwnership(ownerships);
+        assertEquals(ownerships, waitForServerLoadsNotificationAsc(serverLoadNotifications, 2));
+    }
+
+    @Test(timeout = 60000)
+    public void testZkFormatting() throws Exception {
+        final String server = "host/10.0.0.0:31351";
+        final String zkFormattedServer = "host--10.0.0.0:31351";
+        URI uri = new URI("distributedlog-bk://" + zkServers + DLOG_NAMESPACE + "/bknamespace");
+        ZKPlacementStateManager zkPlacementStateManager =
+            new ZKPlacementStateManager(uri, new DistributedLogConfiguration(), NullStatsLogger.INSTANCE);
+        assertEquals(zkFormattedServer, zkPlacementStateManager.serverToZkFormat(server));
+        assertEquals(server, zkPlacementStateManager.zkFormatToServer(zkFormattedServer));
+    }
+
+    @After
+    public void stopZookeeper() throws IOException {
+        zkTestServer.stop();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamManager.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamManager.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamManager.java
new file mode 100644
index 0000000..56e9483
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamManager.java
@@ -0,0 +1,135 @@
+/**
+ * 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.distributedlog.service.stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.distributedlog.DistributedLogConfiguration;
+import org.apache.distributedlog.config.DynamicDistributedLogConfiguration;
+import org.apache.distributedlog.namespace.DistributedLogNamespace;
+import org.apache.distributedlog.service.config.StreamConfigProvider;
+import org.apache.distributedlog.service.streamset.Partition;
+import org.apache.distributedlog.service.streamset.StreamPartitionConverter;
+import com.twitter.util.Await;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+
+/**
+ * Test Case for StreamManager.
+ */
+public class TestStreamManager {
+
+    @Rule
+    public TestName testName = new TestName();
+
+    ScheduledExecutorService mockExecutorService = mock(ScheduledExecutorService.class);
+
+    @Test(timeout = 60000)
+    public void testCollectionMethods() throws Exception {
+        Stream mockStream = mock(Stream.class);
+        when(mockStream.getStreamName()).thenReturn("stream1");
+        when(mockStream.getPartition()).thenReturn(new Partition("stream1", 0));
+        StreamFactory mockStreamFactory = mock(StreamFactory.class);
+        StreamPartitionConverter mockPartitionConverter = mock(StreamPartitionConverter.class);
+        StreamConfigProvider mockStreamConfigProvider = mock(StreamConfigProvider.class);
+        when(mockStreamFactory.create(
+                (String) any(),
+                (DynamicDistributedLogConfiguration) any(),
+                (StreamManager) any())).thenReturn(mockStream);
+        StreamManager streamManager = new StreamManagerImpl(
+                "",
+                new DistributedLogConfiguration(),
+                mockExecutorService,
+                mockStreamFactory,
+                mockPartitionConverter,
+                mockStreamConfigProvider,
+                mock(DistributedLogNamespace.class));
+
+        assertFalse(streamManager.isAcquired("stream1"));
+        assertEquals(0, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+
+        streamManager.notifyAcquired(mockStream);
+        assertTrue(streamManager.isAcquired("stream1"));
+        assertEquals(1, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+
+        streamManager.notifyReleased(mockStream);
+        assertFalse(streamManager.isAcquired("stream1"));
+        assertEquals(0, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+
+        streamManager.notifyAcquired(mockStream);
+        assertTrue(streamManager.isAcquired("stream1"));
+        assertEquals(1, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+
+        streamManager.notifyAcquired(mockStream);
+        assertTrue(streamManager.isAcquired("stream1"));
+        assertEquals(1, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+
+        streamManager.notifyReleased(mockStream);
+        assertFalse(streamManager.isAcquired("stream1"));
+        assertEquals(0, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+
+        streamManager.notifyReleased(mockStream);
+        assertFalse(streamManager.isAcquired("stream1"));
+        assertEquals(0, streamManager.numAcquired());
+        assertEquals(0, streamManager.numCached());
+    }
+
+    @Test(timeout = 60000)
+    public void testCreateStream() throws Exception {
+        Stream mockStream = mock(Stream.class);
+        final String streamName = "stream1";
+        when(mockStream.getStreamName()).thenReturn(streamName);
+        StreamFactory mockStreamFactory = mock(StreamFactory.class);
+        StreamPartitionConverter mockPartitionConverter = mock(StreamPartitionConverter.class);
+        StreamConfigProvider mockStreamConfigProvider = mock(StreamConfigProvider.class);
+        when(mockStreamFactory.create(
+            (String) any(),
+            (DynamicDistributedLogConfiguration) any(),
+            (StreamManager) any())
+        ).thenReturn(mockStream);
+        DistributedLogNamespace dlNamespace = mock(DistributedLogNamespace.class);
+        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
+
+        StreamManager streamManager = new StreamManagerImpl(
+                "",
+                new DistributedLogConfiguration(),
+                executorService,
+                mockStreamFactory,
+                mockPartitionConverter,
+                mockStreamConfigProvider,
+                dlNamespace);
+
+        assertTrue(Await.ready(streamManager.createStreamAsync(streamName)).isReturn());
+        verify(dlNamespace).createLog(streamName);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/c44e0278/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamOp.java
----------------------------------------------------------------------
diff --git a/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamOp.java b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamOp.java
new file mode 100644
index 0000000..a18fda1
--- /dev/null
+++ b/distributedlog-proxy-server/src/test/java/org/apache/distributedlog/service/stream/TestStreamOp.java
@@ -0,0 +1,95 @@
+/**
+ * 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.distributedlog.service.stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.distributedlog.AsyncLogWriter;
+import org.apache.distributedlog.DLSN;
+import org.apache.distributedlog.LogRecord;
+import org.apache.distributedlog.acl.DefaultAccessControlManager;
+import org.apache.distributedlog.exceptions.InternalServerException;
+import org.apache.distributedlog.service.ResponseUtils;
+import org.apache.distributedlog.service.config.ServerConfiguration;
+import org.apache.distributedlog.service.streamset.IdentityStreamPartitionConverter;
+import org.apache.distributedlog.thrift.service.StatusCode;
+import org.apache.distributedlog.thrift.service.WriteResponse;
+import org.apache.distributedlog.util.Sequencer;
+import com.twitter.util.Await;
+import com.twitter.util.Future;
+import java.nio.ByteBuffer;
+import org.apache.bookkeeper.feature.SettableFeature;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+
+/**
+ * Test Case for StreamOps.
+ */
+public class TestStreamOp {
+
+    @Rule
+    public TestName testName = new TestName();
+
+    private WriteOp getWriteOp() {
+        SettableFeature disabledFeature = new SettableFeature("", 0);
+        return new WriteOp("test",
+            ByteBuffer.wrap("test".getBytes()),
+            new NullStatsLogger(),
+            new NullStatsLogger(),
+            new IdentityStreamPartitionConverter(),
+            new ServerConfiguration(),
+            (byte) 0,
+            null,
+            false,
+            disabledFeature,
+            DefaultAccessControlManager.INSTANCE);
+    }
+
+    @Test(timeout = 60000)
+    public void testResponseFailedTwice() throws Exception {
+        WriteOp writeOp = getWriteOp();
+        writeOp.fail(new InternalServerException("test1"));
+        writeOp.fail(new InternalServerException("test2"));
+
+        WriteResponse response = Await.result(writeOp.result());
+        assertEquals(StatusCode.INTERNAL_SERVER_ERROR, response.getHeader().getCode());
+        assertEquals(ResponseUtils.exceptionToHeader(new InternalServerException("test1")), response.getHeader());
+    }
+
+    @Test(timeout = 60000)
+    public void testResponseSucceededThenFailed() throws Exception {
+        AsyncLogWriter writer = mock(AsyncLogWriter.class);
+        when(writer.write((LogRecord) any())).thenReturn(Future.value(new DLSN(1, 2, 3)));
+        when(writer.getStreamName()).thenReturn("test");
+        WriteOp writeOp = getWriteOp();
+        writeOp.execute(writer, new Sequencer() {
+            public long nextId() {
+                return 0;
+            }
+        }, new Object());
+        writeOp.fail(new InternalServerException("test2"));
+
+        WriteResponse response = Await.result(writeOp.result());
+        assertEquals(StatusCode.SUCCESS, response.getHeader().getCode());
+    }
+}


Mime
View raw message