Refactoring the ZooKeeper server to the blur-util project so that a unit test setup to have random ports.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/e0c73d4a Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/e0c73d4a Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/e0c73d4a Branch: refs/heads/master Commit: e0c73d4a82f0acdd228bb0f89e314aeba5c29869 Parents: f5a3799 Author: Aaron McCurry <[email protected]> Authored: Sat Sep 21 22:13:49 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Sat Sep 21 22:13:49 2013 -0400 ---------------------------------------------------------------------- .../test/java/org/apache/blur/MiniCluster.java | 143 ++------------ .../apache/blur/zookeeper/ZkMiniCluster.java | 192 +++++++++++++++++++ .../org/apache/blur/zookeeper/ZkUtilsTest.java | 27 ++- 3 files changed, 216 insertions(+), 146 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e0c73d4a/blur-core/src/test/java/org/apache/blur/MiniCluster.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/MiniCluster.java b/blur-core/src/test/java/org/apache/blur/MiniCluster.java index f99f2c7..01b34a4 100644 --- a/blur-core/src/test/java/org/apache/blur/MiniCluster.java +++ b/blur-core/src/test/java/org/apache/blur/MiniCluster.java @@ -28,7 +28,6 @@ import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; import java.io.File; import java.io.IOException; import java.lang.reflect.Field; -import java.net.InetSocketAddress; import java.net.URI; import java.util.ArrayList; import java.util.List; @@ -61,6 +60,7 @@ import org.apache.blur.thrift.generated.RowMutation; import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.thrift.util.BlurThriftHelper; import org.apache.blur.utils.BlurUtil; +import org.apache.blur.zookeeper.ZkMiniCluster; import org.apache.blur.zookeeper.ZooKeeperClient; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; @@ -72,19 +72,12 @@ import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; -import org.apache.zookeeper.server.ServerCnxnFactory; -import org.apache.zookeeper.server.ServerConfig; -import org.apache.zookeeper.server.ZooKeeperServerMain; -import org.apache.zookeeper.server.quorum.QuorumPeerConfig; -import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException; public class MiniCluster { private static Log LOG = LogFactory.getLog(MiniCluster.class); private MiniDFSCluster cluster; - private Thread serverThread; - // private String zkConnectionString = "localhost:21810"; - private ZooKeeperServerMainEmbedded zooKeeperServerMain; + private ZkMiniCluster zkMiniCluster = new ZkMiniCluster(); private List<ThriftServer> controllers = new ArrayList<ThriftServer>(); private List<ThriftServer> shards = new ArrayList<ThriftServer>(); @@ -334,136 +327,35 @@ public class MiniCluster { } public String getZkConnectionString() { - return zooKeeperServerMain.getConnectionString(); + return zkMiniCluster.getZkConnectionString(); } public void startZooKeeper(String path) { - startZooKeeper(true, path, false); + zkMiniCluster.startZooKeeper(path); } public void startZooKeeper(String path, boolean randomPort) { - startZooKeeper(true, path, randomPort); + zkMiniCluster.startZooKeeper(path, randomPort); } public void startZooKeeper(boolean format, String path) { - startZooKeeper(format, path, false); + zkMiniCluster.startZooKeeper(format, path); } public void startZooKeeper(boolean format, String path, boolean randomPort) { - Properties properties = new Properties(); - properties.setProperty("tickTime", "2000"); - properties.setProperty("initLimit", "10"); - properties.setProperty("syncLimit", "5"); - - properties.setProperty("clientPort", "21810"); - - startZooKeeper(properties, format, path, randomPort); + zkMiniCluster.startZooKeeper(format, path, randomPort); } public void startZooKeeper(Properties properties, String path) { - startZooKeeper(properties, true, path, false); + zkMiniCluster.startZooKeeper(properties, path); } public void startZooKeeper(Properties properties, String path, boolean randomPort) { - startZooKeeper(properties, true, path, randomPort); - } - - private class ZooKeeperServerMainEmbedded extends ZooKeeperServerMain { - @Override - public void shutdown() { - super.shutdown(); - } - - public String getConnectionString() { - try { - Field field = ZooKeeperServerMain.class.getDeclaredField("cnxnFactory"); - field.setAccessible(true); - ServerCnxnFactory serverCnxnFactory = (ServerCnxnFactory) field.get(this); - InetSocketAddress address = serverCnxnFactory.getLocalAddress(); - if (address == null) { - return null; - } - int localPort = serverCnxnFactory.getLocalPort(); - return address.getAddress().getHostAddress() + ":" + localPort; - } catch (NullPointerException e) { - return null; - } catch (Exception e) { - throw new RuntimeException(e); - } - } + zkMiniCluster.startZooKeeper(properties, path, randomPort); } public void startZooKeeper(final Properties properties, boolean format, String path, final boolean randomPort) { - String realPath = path + "/zk_test"; - properties.setProperty("dataDir", realPath); - final ServerConfig serverConfig = new ServerConfig(); - QuorumPeerConfig config = new QuorumPeerConfig() { - @Override - public InetSocketAddress getClientPortAddress() { - InetSocketAddress clientPortAddress = super.getClientPortAddress(); - if (randomPort) { - return randomPort(clientPortAddress); - } - return clientPortAddress; - } - - private InetSocketAddress randomPort(InetSocketAddress clientPortAddress) { - return new InetSocketAddress(clientPortAddress.getAddress(), 0); - } - }; - try { - config.parseProperties(properties); - } catch (IOException e) { - LOG.error(e); - throw new RuntimeException(e); - } catch (ConfigException e) { - LOG.error(e); - throw new RuntimeException(e); - } - serverConfig.readFrom(config); - rm(new File(realPath)); - serverThread = new Thread(new Runnable() { - - @Override - public void run() { - try { - zooKeeperServerMain = new ZooKeeperServerMainEmbedded(); - zooKeeperServerMain.runFromConfig(serverConfig); - } catch (IOException e) { - LOG.error(e); - } - } - }); - serverThread.start(); - long s = System.nanoTime(); - while (s + 10000000000L > System.nanoTime()) { - try { - Thread.sleep(50); - } catch (InterruptedException e) { - LOG.error(e); - throw new RuntimeException(e); - } - try { - String zkConnectionString = getZkConnectionString(); - if (zkConnectionString == null) { - continue; - } - ZooKeeper zk = new ZooKeeper(getZkConnectionString(), 30000, new Watcher() { - @Override - public void process(WatchedEvent event) { - - } - }); - zk.close(); - break; - } catch (IOException e) { - LOG.error(e); - throw new RuntimeException(e); - } catch (InterruptedException e) { - LOG.error(e); - throw new RuntimeException(e); - } - } + zkMiniCluster.startZooKeeper(properties, format, path, randomPort); } public URI getFileSystemUri() throws IOException { @@ -507,7 +399,7 @@ public class MiniCluster { } public void shutdownZooKeeper() { - zooKeeperServerMain.shutdown(); + zkMiniCluster.shutdownZooKeeper(); } public void shutdownDfs() { @@ -571,17 +463,4 @@ public class MiniCluster { throw new RuntimeException(e); } } - - private static void rm(File file) { - if (!file.exists()) { - return; - } - if (file.isDirectory()) { - for (File f : file.listFiles()) { - rm(f); - } - } - file.delete(); - } - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e0c73d4a/blur-util/src/test/java/org/apache/blur/zookeeper/ZkMiniCluster.java ---------------------------------------------------------------------- diff --git a/blur-util/src/test/java/org/apache/blur/zookeeper/ZkMiniCluster.java b/blur-util/src/test/java/org/apache/blur/zookeeper/ZkMiniCluster.java new file mode 100644 index 0000000..7c4ecea --- /dev/null +++ b/blur-util/src/test/java/org/apache/blur/zookeeper/ZkMiniCluster.java @@ -0,0 +1,192 @@ +package org.apache.blur.zookeeper; + +/** + * 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. + */ + +import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.util.Properties; + +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.server.ServerCnxnFactory; +import org.apache.zookeeper.server.ServerConfig; +import org.apache.zookeeper.server.ZooKeeperServerMain; +import org.apache.zookeeper.server.quorum.QuorumPeerConfig; +import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException; + +public class ZkMiniCluster { + + private static Log LOG = LogFactory.getLog(ZkMiniCluster.class); + private Thread serverThread; + private ZooKeeperServerMainEmbedded zooKeeperServerMain; + + public String getZkConnectionString() { + return zooKeeperServerMain.getConnectionString(); + } + + public void startZooKeeper(String path) { + startZooKeeper(true, path, false); + } + + public void startZooKeeper(String path, boolean randomPort) { + startZooKeeper(true, path, randomPort); + } + + public void startZooKeeper(boolean format, String path) { + startZooKeeper(format, path, false); + } + + public void startZooKeeper(boolean format, String path, boolean randomPort) { + Properties properties = new Properties(); + properties.setProperty("tickTime", "2000"); + properties.setProperty("initLimit", "10"); + properties.setProperty("syncLimit", "5"); + + properties.setProperty("clientPort", "21810"); + + startZooKeeper(properties, format, path, randomPort); + } + + public void startZooKeeper(Properties properties, String path) { + startZooKeeper(properties, true, path, false); + } + + public void startZooKeeper(Properties properties, String path, boolean randomPort) { + startZooKeeper(properties, true, path, randomPort); + } + + private class ZooKeeperServerMainEmbedded extends ZooKeeperServerMain { + @Override + public void shutdown() { + super.shutdown(); + } + + public String getConnectionString() { + try { + Field field = ZooKeeperServerMain.class.getDeclaredField("cnxnFactory"); + field.setAccessible(true); + ServerCnxnFactory serverCnxnFactory = (ServerCnxnFactory) field.get(this); + InetSocketAddress address = serverCnxnFactory.getLocalAddress(); + if (address == null) { + return null; + } + int localPort = serverCnxnFactory.getLocalPort(); + return address.getAddress().getHostAddress() + ":" + localPort; + } catch (NullPointerException e) { + return null; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + } + + public void startZooKeeper(final Properties properties, boolean format, String path, final boolean randomPort) { + String realPath = path + "/zk_test"; + properties.setProperty("dataDir", realPath); + final ServerConfig serverConfig = new ServerConfig(); + QuorumPeerConfig config = new QuorumPeerConfig() { + @Override + public InetSocketAddress getClientPortAddress() { + InetSocketAddress clientPortAddress = super.getClientPortAddress(); + if (randomPort) { + return randomPort(clientPortAddress); + } + return clientPortAddress; + } + + private InetSocketAddress randomPort(InetSocketAddress clientPortAddress) { + return new InetSocketAddress(clientPortAddress.getAddress(), 0); + } + }; + try { + config.parseProperties(properties); + } catch (IOException e) { + LOG.error(e); + throw new RuntimeException(e); + } catch (ConfigException e) { + LOG.error(e); + throw new RuntimeException(e); + } + serverConfig.readFrom(config); + rm(new File(realPath)); + serverThread = new Thread(new Runnable() { + + @Override + public void run() { + try { + zooKeeperServerMain = new ZooKeeperServerMainEmbedded(); + zooKeeperServerMain.runFromConfig(serverConfig); + } catch (IOException e) { + LOG.error(e); + } + } + }); + serverThread.start(); + long s = System.nanoTime(); + while (s + 10000000000L > System.nanoTime()) { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + LOG.error(e); + throw new RuntimeException(e); + } + try { + String zkConnectionString = getZkConnectionString(); + if (zkConnectionString == null) { + continue; + } + ZooKeeper zk = new ZooKeeper(getZkConnectionString(), 30000, new Watcher() { + @Override + public void process(WatchedEvent event) { + + } + }); + zk.close(); + break; + } catch (IOException e) { + LOG.error(e); + throw new RuntimeException(e); + } catch (InterruptedException e) { + LOG.error(e); + throw new RuntimeException(e); + } + } + } + + public void shutdownZooKeeper() { + zooKeeperServerMain.shutdown(); + } + + private static void rm(File file) { + if (!file.exists()) { + return; + } + if (file.isDirectory()) { + for (File f : file.listFiles()) { + rm(f); + } + } + file.delete(); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e0c73d4a/blur-util/src/test/java/org/apache/blur/zookeeper/ZkUtilsTest.java ---------------------------------------------------------------------- diff --git a/blur-util/src/test/java/org/apache/blur/zookeeper/ZkUtilsTest.java b/blur-util/src/test/java/org/apache/blur/zookeeper/ZkUtilsTest.java index 157ffa4..6506e64 100644 --- a/blur-util/src/test/java/org/apache/blur/zookeeper/ZkUtilsTest.java +++ b/blur-util/src/test/java/org/apache/blur/zookeeper/ZkUtilsTest.java @@ -21,45 +21,44 @@ import static org.hamcrest.CoreMatchers.notNullValue; import static org.hamcrest.CoreMatchers.nullValue; import static org.junit.Assert.assertThat; +import java.io.File; import java.io.IOException; -import org.apache.blur.zookeeper.ZkUtils; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; -import org.apache.zookeeper.server.quorum.QuorumPeerMain; import org.junit.After; +import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; - public class ZkUtilsTest { private static final int ANY_VERSION = -1; - private static Thread _serverThread; - private ZooKeeper _zooKeeper; + private static ZkMiniCluster _zkMiniCluster; @BeforeClass public static void setupZookeeper() { - _serverThread = new Thread(new Runnable() { - @Override - public void run() { - QuorumPeerMain.main(new String[] { "./src/test/resources/zoo.cfg" }); - } - }); - _serverThread.setDaemon(true); - _serverThread.start(); + _zkMiniCluster = new ZkMiniCluster(); + _zkMiniCluster.startZooKeeper(new File("target/ZkUtilsTest").getAbsolutePath(), true); + } + + @AfterClass + public static void tearDownZookeeper() { + _zkMiniCluster.shutdownZooKeeper(); } + private ZooKeeper _zooKeeper; + @Before public void setUp() throws IOException, InterruptedException { final Object lock = new Object(); synchronized (lock) { - _zooKeeper = new ZooKeeper("127.0.0.1:10101", 10000, new Watcher() { + _zooKeeper = new ZooKeeper(_zkMiniCluster.getZkConnectionString(), 10000, new Watcher() { @Override public void process(WatchedEvent event) { synchronized (lock) {
