Repository: incubator-blur Updated Branches: refs/heads/master 2e57e8d81 -> 64dc0ae2d
BLUR-342 support fully 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/64dc0ae2 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/64dc0ae2 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/64dc0ae2 Branch: refs/heads/master Commit: 64dc0ae2d4626fc75bfeff3c44d628aa2161f3b1 Parents: 2e57e8d Author: Tim <[email protected]> Authored: Thu Jul 31 10:41:24 2014 -0400 Committer: Tim <[email protected]> Committed: Thu Jul 31 10:41:24 2014 -0400 ---------------------------------------------------------------------- .../clusterstatus/ZookeeperClusterStatus.java | 14 +- .../clusterstatus/ZookeeperPathConstants.java | 74 ------ .../indexserver/DistributedIndexServer.java | 36 +-- .../indexserver/DistributedLayoutFactory.java | 3 +- .../indexserver/DistributedLayoutManager.java | 246 ------------------- .../MasterBasedDistributedLayoutFactory.java | 12 +- .../blur/thrift/BlurControllerServer.java | 17 +- .../java/org/apache/blur/thrift/Shutdown.java | 2 +- .../blur/thrift/ThriftBlurControllerServer.java | 42 ++-- .../blur/thrift/ThriftBlurShardServer.java | 55 +++-- .../java/org/apache/blur/utils/BlurUtil.java | 2 +- .../test/java/org/apache/blur/MiniCluster.java | 14 +- .../ZookeeperClusterStatusTest.java | 1 + .../DistributedLayoutManagerTest.java | 116 --------- ...MasterBasedDistributedLayoutFactoryTest.java | 44 ++-- .../main/java/org/apache/blur/shell/Main.java | 69 ++---- .../java/org/apache/blur/thrift/BlurClient.java | 45 ++++ .../java/org/apache/blur/BlurConfiguration.java | 8 + .../org/apache/blur/utils/BlurConstants.java | 3 +- .../blur/zookeeper/ZookeeperPathConstants.java | 74 ++++++ .../src/main/resources/blur-default.properties | 8 +- 21 files changed, 243 insertions(+), 642 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java index df45b73..0a9a57d 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java +++ b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java @@ -48,6 +48,7 @@ import org.apache.blur.zookeeper.WatchChildren.OnChange; import org.apache.blur.zookeeper.WatchNodeData; import org.apache.blur.zookeeper.ZkUtils; import org.apache.blur.zookeeper.ZooKeeperLockManager; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; @@ -244,18 +245,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { @Override public List<String> getControllerServerList() { - long s = System.nanoTime(); - try { - checkIfOpen(); - return _zk.getChildren(ZookeeperPathConstants.getControllersPath(), false); - } catch (KeeperException e) { - throw new RuntimeException(e); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } finally { - long e = System.nanoTime(); - LOG.debug("trace getControllerServerList [" + (e - s) / 1000000.0 + " ms]"); - } + return getOnlineControllerList(); } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperPathConstants.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperPathConstants.java b/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperPathConstants.java deleted file mode 100644 index a8099c1..0000000 --- a/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperPathConstants.java +++ /dev/null @@ -1,74 +0,0 @@ -package org.apache.blur.manager.clusterstatus; - -/** - * 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. - */ - -public class ZookeeperPathConstants { - - public static String getBasePath() { - return "/blur"; - } - - public static String getClusterPath(String cluster) { - return getClustersPath() + "/" + cluster; - } - - public static String getClustersPath() { - return getBasePath() + "/clusters"; - } - - public static String getOnlineControllersPath() { - return getBasePath() + "/online-controller-nodes"; - } - - public static String getControllersPath() { - return getBasePath() + "/controller-nodes"; - } - - public static String getOnlineShardsPath(String cluster) { - return getClusterPath(cluster) + "/online-nodes"; - } - - public static String getTablesPath(String cluster) { - return getClusterPath(cluster) + "/tables"; - } - - public static String getTablePath(String cluster, String table) { - return getTablesPath(cluster) + "/" + table; - } - - public static String getSafemodePath(String cluster) { - return getClusterPath(cluster) + "/safemode"; - } - - public static String getShutdownPath(String cluster) { - return getClusterPath(cluster) + "/shutdown"; - } - - public static String getRegisteredShardsPath(String cluster) { - return getClusterPath(cluster) + "/registered-nodes"; - } - - public static String getShardLayoutPath(String cluster) { - return getClusterPath(cluster) + "/layout"; - } - - public static String getShardLayoutPathLocks(String cluster) { - return getShardLayoutPath(cluster) + "/locks"; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java index 59d3f68..46c0d38 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java @@ -42,7 +42,6 @@ import org.apache.blur.log.LogFactory; import org.apache.blur.manager.BlurFilterCache; import org.apache.blur.manager.clusterstatus.ClusterStatus; import org.apache.blur.manager.clusterstatus.ClusterStatus.Action; -import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.manager.writer.BlurIndexCloser; import org.apache.blur.manager.writer.BlurIndexReadOnly; @@ -60,6 +59,7 @@ import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.utils.BlurUtil; import org.apache.blur.zookeeper.WatchChildren; import org.apache.blur.zookeeper.WatchChildren.OnChange; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.lucene.index.IndexReader; @@ -140,7 +140,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { _internalSearchThreads = internalSearchThreads; _warmupThreads = warmupThreads; _blockCacheDirectoryFactory = blockCacheDirectoryFactory; - _distributedLayoutFactory = distributedLayoutFactory == null ? getDefaultLayoutFactory() : distributedLayoutFactory; + _distributedLayoutFactory = distributedLayoutFactory; _closer.register(_shardStateManager); @@ -204,33 +204,6 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } } - private DistributedLayoutFactory getDefaultLayoutFactory() { - return new DistributedLayoutFactory() { - - @Override - public DistributedLayout createDistributedLayout(String table, List<String> shardList, - List<String> shardServerList, List<String> offlineShardServers) { - DistributedLayoutManager layoutManager = new DistributedLayoutManager(); - layoutManager.setNodes(shardServerList); - layoutManager.setNodesOffline(offlineShardServers); - layoutManager.setShards(shardList); - layoutManager.init(); - return layoutManager; - } - - @Override - public DistributedLayout readCurrentLayout(String table) { - throw new RuntimeException("Not implemented"); - } - - @Override - public Map<String, ?> getLayoutCache() { - throw new RuntimeException("Not implemented"); - } - - }; - } - public static interface ReleaseReader { void release() throws IOException; } @@ -677,8 +650,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { if (cluster == null) { throw new RuntimeException("Table [" + table + "] is not found."); } - List<String> shardServerList = _clusterStatus.getShardServerList(cluster); - List<String> offlineShardServers = new ArrayList<String>(_clusterStatus.getOfflineShardServers(false, cluster)); + List<String> onlineShardServerList = _clusterStatus.getOnlineShardServers(false, cluster); List<String> shardList = getShardList(table); String shutdownPath = ZookeeperPathConstants.getShutdownPath(cluster); @@ -688,7 +660,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } DistributedLayout layoutManager = _distributedLayoutFactory.createDistributedLayout(table, shardList, - shardServerList, offlineShardServers); + onlineShardServerList); Map<String, String> layout = layoutManager.getLayout(); String nodeName = getNodeName(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutFactory.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutFactory.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutFactory.java index a6e3de0..9b9cd36 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutFactory.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutFactory.java @@ -21,8 +21,7 @@ import java.util.Map; public interface DistributedLayoutFactory { - DistributedLayout createDistributedLayout(String table, List<String> shardList, List<String> shardServerList, - List<String> offlineShardServers); + DistributedLayout createDistributedLayout(String table, List<String> shardList, List<String> onlineShardServerList); DistributedLayout readCurrentLayout(String table); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutManager.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutManager.java deleted file mode 100644 index f5b928d..0000000 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedLayoutManager.java +++ /dev/null @@ -1,246 +0,0 @@ -package org.apache.blur.manager.indexserver; - -/** - * 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.util.ArrayList; -import java.util.Collection; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.SortedSet; -import java.util.TreeMap; -import java.util.TreeSet; -import java.util.Map.Entry; - -public class DistributedLayoutManager implements DistributedLayout { - - private static final SortedSet<String> EMPTY_SORTED_SET = new TreeSet<String>(); - - private SortedSet<String> nodes = EMPTY_SORTED_SET; - private Set<String> nodesOffline = EMPTY_SORTED_SET; - private SortedSet<String> shards = EMPTY_SORTED_SET; - private List<String> nodeList = new ArrayList<String>(); - private Map<String, String> cache = new TreeMap<String, String>(); - - public DistributedLayoutManager init() { - if (nodesOffline.equals(nodes) || nodes.isEmpty()) { - cache = getLockedMap(new TreeMap<String, String>()); - return this; - } - Map<String, String> mappings = new TreeMap<String, String>(); - SortedSet<String> moveBecauseOfDownNodes = new TreeSet<String>(); - int nodeListSize = nodeList.size(); - int nodeCount = getStartingPoint(); - for (String shard : shards) { - String node = nodeList.get(nodeCount); - mappings.put(shard, node); - if (nodesOffline.contains(node)) { - moveBecauseOfDownNodes.add(shard); - } - nodeCount++; - if (nodeCount >= nodeListSize) { - nodeCount = 0; - } - } - for (String shard : moveBecauseOfDownNodes) { - String node = nodeList.get(nodeCount); - while (isOffline(node)) { - nodeCount++; - if (nodeCount >= nodeListSize) { - nodeCount = 0; - } - node = nodeList.get(nodeCount); - } - mappings.put(shard, node); - nodeCount++; - if (nodeCount >= nodeListSize) { - nodeCount = 0; - } - } - cache = getLockedMap(mappings); - return this; - } - - private int getStartingPoint() { - int size = nodes.size(); - int hash = 37; - for (String node : nodes) { - hash += node.hashCode() * 17; - } - return Math.abs(hash % size); - } - - @Override - public Map<String, String> getLayout() { - return cache; - } - - private boolean isOffline(String node) { - return nodesOffline.contains(node); - } - - public Collection<String> getNodes() { - return new TreeSet<String>(nodes); - } - - public void setNodes(Collection<String> v) { - this.nodes = new TreeSet<String>(v); - this.nodeList = new ArrayList<String>(nodes); - } - - public Collection<String> getShards() { - return new TreeSet<String>(shards); - } - - public void setShards(Collection<String> v) { - this.shards = new TreeSet<String>(v); - } - - public Collection<String> getNodesOffline() { - return new TreeSet<String>(nodesOffline); - } - - public void setNodesOffline(Collection<String> v) { - this.nodesOffline = new HashSet<String>(v); - } - - private Map<String, String> getLockedMap(final Map<String, String> map) { - final Set<Entry<String, String>> entrySet = wrapEntries(map.entrySet()); - return new Map<String, String>() { - - @Override - public boolean containsKey(Object key) { - return map.containsKey(key); - } - - @Override - public boolean containsValue(Object value) { - return map.containsValue(value); - } - - @Override - public Set<java.util.Map.Entry<String, String>> entrySet() { - return entrySet; - } - - @Override - public String get(Object key) { - return map.get(key); - } - - @Override - public boolean isEmpty() { - return map.isEmpty(); - } - - @Override - public Set<String> keySet() { - return new TreeSet<String>(map.keySet()); - } - - @Override - public int size() { - return map.size(); - } - - @Override - public Collection<String> values() { - return new TreeSet<String>(map.values()); - } - - @Override - public void clear() { - throw new RuntimeException("read only"); - } - - @Override - public String put(String key, String value) { - throw new RuntimeException("read only"); - } - - @Override - public void putAll(Map<? extends String, ? extends String> m) { - throw new RuntimeException("read only"); - } - - @Override - public String remove(Object key) { - throw new RuntimeException("read only"); - } - - @Override - public String toString() { - return map.toString(); - } - - @Override - public boolean equals(Object obj) { - return map.equals(obj); - } - - @Override - public int hashCode() { - return map.hashCode(); - } - }; - } - - private Set<Entry<String, String>> wrapEntries(Set<Entry<String, String>> entrySet) { - Set<Entry<String, String>> result = new HashSet<Entry<String, String>>(); - for (Entry<String, String> entry : entrySet) { - result.add(wrapEntry(entry)); - } - return result; - } - - private Entry<String, String> wrapEntry(final Entry<String, String> entry) { - return new Entry<String, String>() { - - @Override - public String setValue(String value) { - throw new RuntimeException("read only"); - } - - @Override - public String getValue() { - return entry.getValue(); - } - - @Override - public String getKey() { - return entry.getKey(); - } - - @Override - public String toString() { - return entry.toString(); - } - - @Override - public boolean equals(Object obj) { - return entry.equals(obj); - } - - @Override - public int hashCode() { - return entry.hashCode(); - } - }; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactory.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactory.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactory.java index 319ccf2..6fff994 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactory.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactory.java @@ -38,9 +38,9 @@ import java.util.concurrent.ConcurrentMap; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; -import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.zookeeper.ZkUtils; import org.apache.blur.zookeeper.ZooKeeperLockManager; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs.Ids; @@ -105,11 +105,10 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac } @Override - public DistributedLayout createDistributedLayout(String table, List<String> shardList, List<String> shardServerList, - List<String> offlineShardServers) { + public DistributedLayout createDistributedLayout(String table, List<String> shardList, List<String> onlineShardServerList) { LOG.info("Creating layout for table [{0}]", table); MasterBasedDistributedLayout layout = _cachedLayoutMap.get(table); - List<String> onlineShardServerList = getOnlineShardServerList(shardServerList, offlineShardServers); + if (layout == null || layout.isOutOfDate(shardList, onlineShardServerList)) { LOG.info("Layout out of date, recalculating for table [{0}].", table); MasterBasedDistributedLayout newLayout = newLayout(table, shardList, onlineShardServerList); @@ -121,11 +120,6 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac } } - private List<String> getOnlineShardServerList(List<String> shardServerList, List<String> offlineShardServers) { - List<String> list = new ArrayList<String>(shardServerList); - list.removeAll(offlineShardServers); - return list; - } private MasterBasedDistributedLayout newLayout(String table, List<String> shardList, List<String> onlineShardServerList) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java index a1cec15..e378c44 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java @@ -53,7 +53,6 @@ import org.apache.blur.log.LogFactory; import org.apache.blur.manager.BlurPartitioner; import org.apache.blur.manager.BlurQueryChecker; import org.apache.blur.manager.IndexManager; -import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.manager.indexserver.DistributedLayout; import org.apache.blur.manager.indexserver.DistributedLayoutFactory; import org.apache.blur.manager.indexserver.DistributedLayoutFactoryImpl; @@ -100,6 +99,7 @@ import org.apache.blur.utils.ForkJoin; import org.apache.blur.utils.ForkJoin.Merger; import org.apache.blur.utils.ForkJoin.ParallelCall; import org.apache.blur.zookeeper.WatchChildren; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.blur.zookeeper.WatchChildren.OnChange; import org.apache.blur.zookeeper.WatchNodeExistance; import org.apache.zookeeper.CreateMode; @@ -273,7 +273,6 @@ public class BlurControllerServer extends TableAdmin implements Iface { private void setupZookeeper() throws KeeperException, InterruptedException { BlurUtil.createIfMissing(_zookeeper, "/blur"); BlurUtil.createIfMissing(_zookeeper, ZookeeperPathConstants.getOnlineControllersPath()); - BlurUtil.createIfMissing(_zookeeper, ZookeeperPathConstants.getControllersPath()); BlurUtil.createIfMissing(_zookeeper, ZookeeperPathConstants.getClustersPath()); } @@ -387,20 +386,6 @@ public class BlurControllerServer extends TableAdmin implements Iface { } private void registerMyself() { - // Register Node - try { - String controllerPath = ZookeeperPathConstants.getControllersPath() + "/" + _nodeName; - if (_zookeeper.exists(controllerPath, false) == null) { - // Don't set the version for the registered nodes but only to the online - // nodes. - _zookeeper.create(controllerPath, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - } - } catch (KeeperException e) { - throw new RuntimeException(e); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - // Wait for other instances (named the same name) to die try { String version = BlurUtil.getVersion(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/thrift/Shutdown.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/Shutdown.java b/blur-core/src/main/java/org/apache/blur/thrift/Shutdown.java index 5f082e0..737794b 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/Shutdown.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/Shutdown.java @@ -25,8 +25,8 @@ import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT_DEFAULT import java.io.IOException; import org.apache.blur.BlurConfiguration; -import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.zookeeper.ZkUtils; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs.Ids; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java index 76897f6..7814582 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java @@ -39,6 +39,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_GUI_CONTROLLER_PORT; import static org.apache.blur.utils.BlurConstants.BLUR_HTTP_STATUS_RUNNING_PORT; import static org.apache.blur.utils.BlurConstants.BLUR_MAX_RECORDS_PER_ROW_FETCH_REQUEST; import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_MAX_FRAME_SIZE; +import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT_DEFAULT; @@ -84,7 +85,7 @@ public class ThriftBlurControllerServer extends ThriftServer { ReporterSetup.setupReporters(configuration); MemoryReporter.enable(); setupJvmMetrics(); - ThriftServer server = createServer(serverIndex, configuration, false); + ThriftServer server = createServer(serverIndex, configuration); server.start(); } catch (Throwable t) { t.printStackTrace(); @@ -92,25 +93,25 @@ public class ThriftBlurControllerServer extends ThriftServer { } } - public static ThriftServer createServer(int serverIndex, BlurConfiguration configuration, boolean randomPort) - throws Exception { + public static ThriftServer createServer(int serverIndex, BlurConfiguration configuration) throws Exception { Thread.setDefaultUncaughtExceptionHandler(new SimpleUncaughtExceptionHandler()); String bindAddress = configuration.get(BLUR_CONTROLLER_BIND_ADDRESS); - int bindPort = configuration.getInt(BLUR_CONTROLLER_BIND_PORT, -1); - bindPort += serverIndex; - if (randomPort) { - bindPort = 0; + int configBindPort = configuration.getInt(BLUR_CONTROLLER_BIND_PORT, -1); + int instanceBindPort = configBindPort + serverIndex; + if (configBindPort == 0) { + instanceBindPort = 0; } - TNonblockingServerSocket tNonblockingServerSocket = ThriftServer.getTNonblockingServerSocket(bindAddress, bindPort); - if (randomPort) { - bindPort = tNonblockingServerSocket.getServerSocket().getLocalPort(); + TNonblockingServerSocket tNonblockingServerSocket = ThriftServer.getTNonblockingServerSocket(bindAddress, + instanceBindPort); + if (configBindPort == 0) { + instanceBindPort = tNonblockingServerSocket.getServerSocket().getLocalPort(); } - LOG.info("Controller Server using index [{0}] bind address [{1}] random port assignment [{2}]", serverIndex, - bindAddress + ":" + bindPort, randomPort); + LOG.info("Controller Server using index [{0}] bind address [{1}]", serverIndex, bindAddress + ":" + + instanceBindPort); String nodeName = getNodeName(configuration, BLUR_CONTROLLER_HOSTNAME); - nodeName = nodeName + ":" + bindPort; + nodeName = nodeName + ":" + instanceBindPort; String zkConnectionStr = isEmpty(configuration.get(BLUR_ZOOKEEPER_CONNECTION), BLUR_ZOOKEEPER_CONNECTION); BlurQueryChecker queryChecker = new BlurQueryChecker(configuration); @@ -171,13 +172,18 @@ public class ThriftBlurControllerServer extends ThriftServer { server.setAcceptQueueSizePerThread(configuration.getInt(BLUR_CONTROLLER_THRIFT_ACCEPT_QUEUE_SIZE_PER_THREAD, 4)); server.setMaxReadBufferBytes(configuration.getLong(BLUR_CONTROLLER_THRIFT_MAX_READ_BUFFER_BYTES, Long.MAX_VALUE)); server.setSelectorThreads(configuration.getInt(BLUR_CONTROLLER_THRIFT_SELECTOR_THREADS, 2)); - server.setMaxFrameSize(configuration.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, 16384000)); + server.setMaxFrameSize(configuration.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE)); + + int configGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_CONTROLLER_PORT)); + int instanceGuiPort = configGuiPort + serverIndex; + + if (configGuiPort == 0) { + instanceGuiPort = 0; + } - int baseGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_CONTROLLER_PORT)); final HttpJettyServer httpServer; - if (baseGuiPort >= 0) { - int webServerPort = baseGuiPort + serverIndex; - httpServer = new HttpJettyServer(HttpJettyServer.class, webServerPort); + if (configGuiPort >= 0) { + httpServer = new HttpJettyServer(HttpJettyServer.class, instanceGuiPort); int port = httpServer.getLocalPort(); configuration.setInt(BLUR_HTTP_STATUS_RUNNING_PORT, port); } else { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index cd39485..195eb5e 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -46,6 +46,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_SELECTOR_THR import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_WARMUP_DISABLED; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_WARMUP_THREAD_COUNT; import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_MAX_FRAME_SIZE; +import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT_DEFAULT; @@ -121,7 +122,7 @@ public class ThriftBlurShardServer extends ThriftServer { setupJvmMetrics(); // make this configurable GCWatcher.init(0.75); - ThriftServer server = createServer(serverIndex, configuration, false); + ThriftServer server = createServer(serverIndex, configuration); server.start(); } catch (Throwable t) { t.printStackTrace(); @@ -129,32 +130,21 @@ public class ThriftBlurShardServer extends ThriftServer { } } - public static ThriftServer createServer(int serverIndex, BlurConfiguration configuration, boolean randomPort) - throws Exception { + public static ThriftServer createServer(int serverIndex, BlurConfiguration configuration) throws Exception { Configuration config = new Configuration(); TableContext.setSystemBlurConfiguration(configuration); TableContext.setSystemConfiguration(config); String bindAddress = configuration.get(BLUR_SHARD_BIND_ADDRESS); - int bindPort = configuration.getInt(BLUR_SHARD_BIND_PORT, -1); - bindPort += serverIndex; - if (randomPort) { - bindPort = 0; + int configBindPort = configuration.getInt(BLUR_SHARD_BIND_PORT, -1); + int instanceBindPort = configBindPort + serverIndex; + if (configBindPort == 0) { + instanceBindPort = 0; } - TNonblockingServerSocket tNonblockingServerSocket = ThriftServer.getTNonblockingServerSocket(bindAddress, bindPort); - if (randomPort) { - bindPort = tNonblockingServerSocket.getServerSocket().getLocalPort(); - } - - int baseGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_SHARD_PORT)); - final HttpJettyServer httpServer; - if (baseGuiPort > 0) { - int webServerPort = baseGuiPort + serverIndex; - httpServer = new HttpJettyServer(HttpJettyServer.class, webServerPort); - int port = httpServer.getLocalPort(); - configuration.setInt(BLUR_HTTP_STATUS_RUNNING_PORT, port); - } else { - httpServer = null; + TNonblockingServerSocket tNonblockingServerSocket = ThriftServer.getTNonblockingServerSocket(bindAddress, + instanceBindPort); + if (configBindPort == 0) { + instanceBindPort = tNonblockingServerSocket.getServerSocket().getLocalPort(); } Set<Entry<String, String>> set = configuration.getProperties().entrySet(); @@ -180,11 +170,10 @@ public class ThriftBlurShardServer extends ThriftServer { } else { throw new RuntimeException("Unknown block cache version [" + blockCacheVersion + "] can be [v1,v2]"); } - LOG.info("Shard Server using index [{0}] bind address [{1}] random port assignment [{2}]", serverIndex, bindAddress - + ":" + bindPort, randomPort); + LOG.info("Shard Server using index [{0}] bind address [{1}]", serverIndex, bindAddress + ":" + instanceBindPort); String nodeNameHostName = getNodeName(configuration, BLUR_SHARD_HOSTNAME); - String nodeName = nodeNameHostName + ":" + bindPort; + String nodeName = nodeNameHostName + ":" + instanceBindPort; String zkConnectionStr = isEmpty(configuration.get(BLUR_ZOOKEEPER_CONNECTION), BLUR_ZOOKEEPER_CONNECTION); BlurQueryChecker queryChecker = new BlurQueryChecker(configuration); @@ -258,6 +247,22 @@ public class ThriftBlurShardServer extends ThriftServer { iface = BlurUtil.runWithUser(iface, false); iface = BlurUtil.runTrace(iface, false); iface = BlurUtil.lastChanceErrorHandling(iface, Iface.class); + + int configGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_SHARD_PORT)); + int instanceGuiPort = configGuiPort + serverIndex; + + if (configGuiPort == 0) { + instanceGuiPort = 0; + } + + final HttpJettyServer httpServer; + if (configGuiPort >= 0) { + httpServer = new HttpJettyServer(HttpJettyServer.class, instanceGuiPort); + int port = httpServer.getLocalPort(); + configuration.setInt(BLUR_HTTP_STATUS_RUNNING_PORT, port); + } else { + httpServer = null; + } if (httpServer != null) { WebAppContext context = httpServer.getContext(); context.addServlet(new ServletHolder(new TServlet(new Blur.Processor<Blur.Iface>(iface), @@ -281,7 +286,7 @@ public class ThriftBlurShardServer extends ThriftServer { server.setAcceptQueueSizePerThread(configuration.getInt(BLUR_SHARD_THRIFT_ACCEPT_QUEUE_SIZE_PER_THREAD, 4)); server.setMaxReadBufferBytes(configuration.getLong(BLUR_SHARD_THRIFT_MAX_READ_BUFFER_BYTES, Long.MAX_VALUE)); server.setSelectorThreads(configuration.getInt(BLUR_SHARD_THRIFT_SELECTOR_THREADS, 2)); - server.setMaxFrameSize(config.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, 16384000)); + server.setMaxFrameSize(config.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE)); // This will shutdown the server when the correct path is set in zk BlurShutdown shutdown = new BlurShutdown() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java index 52fd09f..b57573d 100644 --- a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java +++ b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java @@ -68,7 +68,6 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.lucene.search.PrimeDocCache; import org.apache.blur.manager.BlurHighlighter; -import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.manager.results.BlurResultComparator; import org.apache.blur.manager.results.BlurResultIterable; import org.apache.blur.manager.results.BlurResultPeekableIteratorComparator; @@ -104,6 +103,7 @@ import org.apache.blur.trace.Trace; import org.apache.blur.trace.Tracer; import org.apache.blur.user.User; import org.apache.blur.user.UserContext; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.hadoop.conf.Configurable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/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 7e85201..ff58ff3 100644 --- a/blur-core/src/test/java/org/apache/blur/MiniCluster.java +++ b/blur-core/src/test/java/org/apache/blur/MiniCluster.java @@ -24,6 +24,8 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCKCACHE_SLAB_COU import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_HOSTNAME; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SAFEMODEDELAY; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; +import static org.apache.blur.utils.BlurConstants.BLUR_CONTROLLER_BIND_PORT; +import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BIND_PORT; import java.io.File; import java.io.IOException; @@ -42,7 +44,6 @@ import java.util.concurrent.TimeUnit; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; -import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.store.buffer.BufferStore; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransportException; @@ -64,6 +65,7 @@ import org.apache.blur.utils.BlurUtil; import org.apache.blur.utils.MemoryReporter; import org.apache.blur.zookeeper.ZkMiniCluster; import org.apache.blur.zookeeper.ZooKeeperClient; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; @@ -264,9 +266,12 @@ public class MiniCluster { public void startControllers(BlurConfiguration configuration, int num, boolean randomPort) { BlurConfiguration localConf = getBlurConfiguration(configuration); + if(randomPort) { + localConf.setInt(BLUR_CONTROLLER_BIND_PORT, 0); + } for (int i = 0; i < num; i++) { try { - ThriftServer server = ThriftBlurControllerServer.createServer(i, localConf, randomPort); + ThriftServer server = ThriftBlurControllerServer.createServer(i, localConf); controllers.add(server); startServer(server); } catch (Exception e) { @@ -283,6 +288,9 @@ public class MiniCluster { public void startShards(final BlurConfiguration configuration, int num, final boolean randomPort) { final BlurConfiguration localConf = getBlurConfiguration(configuration); + if(randomPort) { + localConf.setInt(BLUR_SHARD_BIND_PORT, 0); + } ExecutorService executorService = Executors.newFixedThreadPool(num); List<Future<ThriftServer>> futures = new ArrayList<Future<ThriftServer>>(); for (int i = 0; i < num; i++) { @@ -290,7 +298,7 @@ public class MiniCluster { futures.add(executorService.submit(new Callable<ThriftServer>() { @Override public ThriftServer call() throws Exception { - return ThriftBlurShardServer.createServer(index, localConf, randomPort); + return ThriftBlurShardServer.createServer(index, localConf); } })); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/test/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatusTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatusTest.java b/blur-core/src/test/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatusTest.java index d5236f8..deb930d 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatusTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatusTest.java @@ -35,6 +35,7 @@ import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.utils.BlurUtil; import org.apache.blur.zookeeper.ZooKeeperClient; import org.apache.blur.zookeeper.ZooKeeperLockManager; +import org.apache.blur.zookeeper.ZookeeperPathConstants; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/test/java/org/apache/blur/manager/indexserver/DistributedLayoutManagerTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/indexserver/DistributedLayoutManagerTest.java b/blur-core/src/test/java/org/apache/blur/manager/indexserver/DistributedLayoutManagerTest.java deleted file mode 100644 index f3136ea..0000000 --- a/blur-core/src/test/java/org/apache/blur/manager/indexserver/DistributedLayoutManagerTest.java +++ /dev/null @@ -1,116 +0,0 @@ -package org.apache.blur.manager.indexserver; - -/** - * 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 static org.junit.Assert.*; - -import java.util.Arrays; -import java.util.Collection; -import java.util.HashSet; -import java.util.Map; -import java.util.TreeSet; - -import org.apache.blur.manager.indexserver.DistributedLayoutManager; -import org.junit.Test; - - -public class DistributedLayoutManagerTest { - - @Test - public void testLayoutManager() { - TreeSet<String> nodes = new TreeSet<String>(); - nodes.add("n1"); - nodes.add("n2"); - nodes.add("n3"); - - TreeSet<String> nodesOffline = new TreeSet<String>(); - nodesOffline.add("n2"); - - TreeSet<String> shards = new TreeSet<String>(); - shards.add("s1"); - shards.add("s2"); - shards.add("s3"); - shards.add("s4"); - shards.add("s5"); - - DistributedLayoutManager layoutManager1 = new DistributedLayoutManager(); - layoutManager1.setNodes(nodes); - layoutManager1.setShards(shards); - layoutManager1.init(); - Map<String, String> layout1 = layoutManager1.getLayout(); - - DistributedLayoutManager layoutManager2 = new DistributedLayoutManager(); - layoutManager2.setNodes(nodes); - layoutManager2.setShards(shards); - layoutManager2.setNodesOffline(nodesOffline); - layoutManager2.init(); - Map<String, String> layout2 = layoutManager2.getLayout(); - - assertEquals(shards, new TreeSet<String>(layout1.keySet())); - assertEquals(nodes, new TreeSet<String>(layout1.values())); - - assertEquals(shards, new TreeSet<String>(layout2.keySet())); - TreeSet<String> nodesOnline = new TreeSet<String>(nodes); - nodesOnline.removeAll(nodesOffline); - assertEquals(nodesOnline, new TreeSet<String>(layout2.values())); - - } - - @Test - public void testLayoutManagerPerformance() { - DistributedLayoutManager perfTest = new DistributedLayoutManager(); - perfTest.setNodes(getTestNodes()); - perfTest.setShards(getTestShards()); - perfTest.setNodesOffline(getTestOfflineNodes()); - perfTest.init(); - int testSize = 100000; - for (int i = 0; i < testSize; i++) { - perfTest.getLayout(); - } - long s = System.nanoTime(); - for (int i = 0; i < testSize; i++) { - perfTest.getLayout(); - } - long e = System.nanoTime(); - double ms = (e - s) / 1000000.0; - System.out.println("Total " + ms); - System.out.println("Per Call " + ms / testSize); - assertTrue(ms < 100); - } - - private static Collection<String> getTestOfflineNodes() { - return Arrays.asList("n13"); - } - - private static Collection<String> getTestShards() { - Collection<String> shards = new HashSet<String>(); - for (int i = 0; i < 701; i++) { - shards.add("s" + i); - } - return shards; - } - - private static Collection<String> getTestNodes() { - Collection<String> nodes = new HashSet<String>(); - for (int i = 0; i < 32; i++) { - nodes.add("n" + i); - } - return nodes; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-core/src/test/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactoryTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactoryTest.java b/blur-core/src/test/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactoryTest.java index e40b190..1f834d3 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactoryTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/indexserver/MasterBasedDistributedLayoutFactoryTest.java @@ -39,6 +39,8 @@ import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; +import com.google.common.collect.Lists; + @SuppressWarnings("unchecked") public class MasterBasedDistributedLayoutFactoryTest { private static String path = "./target/test-zk-MasterBasedDistributedLayoutFactoryTest"; @@ -79,13 +81,12 @@ public class MasterBasedDistributedLayoutFactoryTest { public void testDecreaseInServers() throws IOException, KeeperException, InterruptedException { MasterBasedDistributedLayoutFactory factory = new MasterBasedDistributedLayoutFactory(_zooKeeper, cluster); - List<String> shardList = list("shard-0", "shard-1", "shard-2", "shard-3", "shard-4", "shard-5"); - List<String> shardServerList = list("server-0", "server-1", "server-2", "server-3", "server-4", "server-5"); - List<String> offlineShardServers = list(); - + List<String> shardList = Lists.newArrayList("shard-0", "shard-1", "shard-2", "shard-3", "shard-4", "shard-5"); + List<String> shardServerList = Lists.newArrayList("server-0", "server-1", "server-2", "server-3", "server-4", "server-5"); + String table = "t1"; - DistributedLayout layout1 = factory.createDistributedLayout(table, shardList, shardServerList, offlineShardServers); + DistributedLayout layout1 = factory.createDistributedLayout(table, shardList, shardServerList); Map<String, String> expected1 = map(e("shard-0", "server-0"), e("shard-1", "server-1"), e("shard-2", "server-2"), e("shard-3", "server-3"), e("shard-4", "server-4"), e("shard-5", "server-5")); @@ -93,11 +94,9 @@ public class MasterBasedDistributedLayoutFactoryTest { assertEquals(expected1, actual1); - List<String> newShardServerList = list("server-0", "server-1", "server-2", "server-3"); - List<String> newOfflineShardServers = list("server-4", "server-5"); - - DistributedLayout layout2 = factory.createDistributedLayout(table, shardList, newShardServerList, - newOfflineShardServers); + List<String> newShardServerList = Lists.newArrayList("server-0", "server-1", "server-2", "server-3"); + + DistributedLayout layout2 = factory.createDistributedLayout(table, shardList, newShardServerList); Map<String, String> expected2 = map(e("shard-0", "server-0"), e("shard-1", "server-1"), e("shard-2", "server-2"), e("shard-3", "server-3"), e("shard-4", "server-0"), e("shard-5", "server-1")); @@ -109,13 +108,12 @@ public class MasterBasedDistributedLayoutFactoryTest { public void testIncreaseInServers() throws IOException, KeeperException, InterruptedException { MasterBasedDistributedLayoutFactory factory = new MasterBasedDistributedLayoutFactory(_zooKeeper, cluster); - List<String> shardList = list("shard-0", "shard-1", "shard-2", "shard-3", "shard-4", "shard-5"); - List<String> shardServerList = list("server-0", "server-1", "server-2", "server-3"); - List<String> offlineShardServers = list(); - + List<String> shardList = Lists.newArrayList("shard-0", "shard-1", "shard-2", "shard-3", "shard-4", "shard-5"); + List<String> shardServerList = Lists.newArrayList("server-0", "server-1", "server-2", "server-3"); + String table = "t1"; - DistributedLayout layout1 = factory.createDistributedLayout(table, shardList, shardServerList, offlineShardServers); + DistributedLayout layout1 = factory.createDistributedLayout(table, shardList, shardServerList); Map<String, String> expected1 = map(e("shard-0", "server-0"), e("shard-1", "server-1"), e("shard-2", "server-2"), e("shard-3", "server-3"), e("shard-4", "server-0"), e("shard-5", "server-1")); @@ -123,11 +121,9 @@ public class MasterBasedDistributedLayoutFactoryTest { assertEquals(expected1, actual1); - List<String> newShardServerList = list("server-0", "server-1", "server-2", "server-3", "server-4", "server-5"); - List<String> newOfflineShardServers = list(); - - DistributedLayout layout2 = factory.createDistributedLayout(table, shardList, newShardServerList, - newOfflineShardServers); + List<String> newShardServerList = Lists.newArrayList("server-0", "server-1", "server-2", "server-3", "server-4", "server-5"); + + DistributedLayout layout2 = factory.createDistributedLayout(table, shardList, newShardServerList); Map<String, String> expected2 = map(e("shard-0", "server-4"), e("shard-1", "server-5"), e("shard-2", "server-2"), e("shard-3", "server-3"), e("shard-4", "server-0"), e("shard-5", "server-1")); @@ -148,14 +144,6 @@ public class MasterBasedDistributedLayoutFactoryTest { zooKeeper.delete(storagePath, -1); } - private static List<String> list(String... list) { - List<String> lst = new ArrayList<String>(); - for (String s : list) { - lst.add(s); - } - return lst; - } - private static Map<String, String> map(Entry<String, String>... entries) { Map<String, String> map = new TreeMap<String, String>(); for (Entry<String, String> e : entries) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-shell/src/main/java/org/apache/blur/shell/Main.java ---------------------------------------------------------------------- diff --git a/blur-shell/src/main/java/org/apache/blur/shell/Main.java b/blur-shell/src/main/java/org/apache/blur/shell/Main.java index 808a53d..98c1ab3 100644 --- a/blur-shell/src/main/java/org/apache/blur/shell/Main.java +++ b/blur-shell/src/main/java/org/apache/blur/shell/Main.java @@ -99,7 +99,7 @@ public class Main { private static class CliShellOptions { - private String _controllerConnectionString; + private Iface _client; private boolean _shell; private String[] _args; private String _command; @@ -108,8 +108,8 @@ public class Main { return _shell; } - public String getControllerConnectionString() { - return _controllerConnectionString; + public Iface getClient() { + return _client; } public String getCommand() { @@ -120,8 +120,8 @@ public class Main { return _args; } - public void setControllerConnectionString(String controllerConnectionString) { - this._controllerConnectionString = controllerConnectionString; + public void setClient(Iface client) { + this._client = client; } public void setShell(boolean shell) { @@ -550,12 +550,12 @@ public class Main { } try { - Blur.Iface client = BlurClient.getClient(cliShellOptions.getControllerConnectionString()); + Blur.Iface client = cliShellOptions.getClient(); if (cliShellOptions.isShell()) { ConsoleReader reader = new ConsoleReader(); PrintWriter out = new PrintWriter(reader.getOutput()); setConsoleReader(commands, reader); - setPrompt(client, reader, cliShellOptions.getControllerConnectionString(), out); + setPrompt(client, reader, out); List<Completer> completors = new LinkedList<Completer>(); @@ -621,7 +621,7 @@ public class Main { out.println("Last command took " + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) + "ms"); } } - setPrompt(client, reader, cliShellOptions.getControllerConnectionString(), out); + setPrompt(client, reader, out); } } } finally { @@ -705,18 +705,13 @@ public class Main { builder.put(command.name(), command); } - private static void setPrompt(Iface client, ConsoleReader reader, String connectionStr, PrintWriter out) + private static void setPrompt(Iface client, ConsoleReader reader, PrintWriter out) throws BlurException, TException, CommandException, IOException { List<String> shardClusterList; try { - shardClusterList = BlurClientManager.execute(connectionStr, new BlurCommand<List<String>>() { - @Override - public List<String> call(Client client) throws BlurException, TException { - return client.shardClusterList(); - } - }, 0, 0, 0); - } catch (BadConnectionException e) { - out.println(e.getMessage() + " Connection (" + connectionStr + ")"); + shardClusterList = client.shardClusterList(); + } catch (BlurException e) { + out.println("Unable to retrieve cluster information - " + e.getMessage()); out.flush(); if (debug) { e.printStackTrace(out); @@ -752,13 +747,7 @@ public class Main { private static CliShellOptions getCliShellOptions(String[] args) throws IOException { CliShellOptions cliShellOptions = new CliShellOptions(); if (args.length == 0) { - String controllerConnectionString = loadControllerConnectionString(); - if (controllerConnectionString == null) { - System.err - .println("Could not locate controller connection string in the blu-site.properties file and it was not passed in via command line args."); - return null; - } - cliShellOptions.setControllerConnectionString(controllerConnectionString); + cliShellOptions.setClient(BlurClient.getClient()); cliShellOptions.setShell(true); return cliShellOptions; } else { @@ -774,7 +763,7 @@ public class Main { + message + "]"); return null; } - cliShellOptions.setControllerConnectionString(arg0); + cliShellOptions.setClient(BlurClient.getClient(arg0)); if (args.length > 1) { // there's might be a command after the connection string cliShellOptions.setShell(false); @@ -795,13 +784,7 @@ public class Main { return cliShellOptions; } } else { - String controllerConnectionString = loadControllerConnectionString(); - if (controllerConnectionString == null) { - System.err - .println("Could not locate controller connection string in the blur-site.properties file and it was not passed in via command line args."); - return null; - } - cliShellOptions.setControllerConnectionString(controllerConnectionString); + cliShellOptions.setClient(BlurClient.getClient()); // command was found at arg0 cliShellOptions.setShell(false); cliShellOptions.setArgs(args); @@ -817,26 +800,4 @@ public class Main { } } - private static String loadControllerConnectionString() throws IOException { - StringBuilder builder = new StringBuilder(); - InputStream inputStream = Main.class.getResourceAsStream("/controllers"); - if (inputStream == null) { - return null; - } - BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream)); - String line; - while ((line = bufferedReader.readLine()) != null) { - if (builder.length() != 0) { - builder.append(','); - } - String trim = line.trim(); - if (trim.startsWith("#")) { - continue; - } - builder.append(trim).append(":40010"); - } - bufferedReader.close(); - return builder.toString(); - } - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java index 7d5e1d2..ad4813a 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java @@ -2,6 +2,7 @@ package org.apache.blur.thrift; /** * 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 @@ -16,6 +17,11 @@ package org.apache.blur.thrift; * See the License for the specific language governing permissions and * limitations under the License. */ +import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; +import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT; +import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TIMEOUT_DEFAULT; + +import java.io.IOException; import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; @@ -23,11 +29,19 @@ import java.lang.reflect.Proxy; import java.util.Arrays; import java.util.List; +import org.apache.blur.BlurConfiguration; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thrift.commands.BlurCommand; import org.apache.blur.thrift.generated.Blur.Client; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.zookeeper.ZkUtils; +import org.apache.blur.zookeeper.ZookeeperPathConstants; +import org.apache.commons.lang.StringUtils; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZooKeeper; + + public class BlurClient { @@ -77,6 +91,20 @@ public class BlurClient { } } + + public static Iface getClient() { + try { + return getClient(new BlurConfiguration()); + } catch (IOException e) { + throw new RuntimeException("Unable to load configurations.", e); + } + } + + public static Iface getClient(BlurConfiguration conf) { + List<String> onlineControllers = getOnlineControllers(conf); + + return getClient(StringUtils.join(onlineControllers, ",")); + } /** * Returns a client interface to Blur based on the connectionStr. @@ -123,5 +151,22 @@ public class BlurClient { return (Iface) Proxy.newProxyInstance(Iface.class.getClassLoader(), new Class[] { Iface.class }, new BlurClientInvocationHandler(connections, maxRetries, backOffTime, maxBackOffTime)); } + + private static List<String> getOnlineControllers(BlurConfiguration conf) { + String zkConn = conf.getExpected(BLUR_ZOOKEEPER_CONNECTION); + int zkSessionTimeout = conf.getInt(BLUR_ZOOKEEPER_TIMEOUT, BLUR_ZOOKEEPER_TIMEOUT_DEFAULT); + + ZooKeeper zkClient = null; + try { + zkClient = ZkUtils.newZooKeeper(zkConn, zkSessionTimeout); + return zkClient.getChildren(ZookeeperPathConstants.getOnlineControllersPath(), false); + } catch (KeeperException e) { + throw new RuntimeException("Error communicating with Zookeeper", e); + } catch (InterruptedException e) { + throw new RuntimeException("Error communicating with Zookeeper", e); + } catch (IOException e) { + throw new RuntimeException("Unable to initialize Zookeeper", e); + } + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-util/src/main/java/org/apache/blur/BlurConfiguration.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/BlurConfiguration.java b/blur-util/src/main/java/org/apache/blur/BlurConfiguration.java index faeb11e..c6b256f 100644 --- a/blur-util/src/main/java/org/apache/blur/BlurConfiguration.java +++ b/blur-util/src/main/java/org/apache/blur/BlurConfiguration.java @@ -88,6 +88,14 @@ public class BlurConfiguration implements Cloneable { public String get(String name) { return get(name, null); } + + public String getExpected(String name) { + String val = get(name); + if (val == null || val.trim().isEmpty()) { + throw new IllegalArgumentException("Property [" + name + "] is missing or blank."); + } + return val; + } public String get(String name, String defaultValue) { String property = _properties.getProperty(name, defaultValue); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java index e8d406e..66fbe8d 100644 --- a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java +++ b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java @@ -145,7 +145,8 @@ public class BlurConstants { public static final String BLUR_LUCENE_FST_BYTEARRAY_FACTORY = "blur.lucene.fst.bytearray.factory"; public static final String BLUR_THRIFT_MAX_FRAME_SIZE = "blur.thrift.max.frame.size"; - + public static final int BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE = 16384000; + public static final String BLUR_SHARD_FILTERED_SERVER_CLASS = "blur.shard.filtered.server.class"; public static final String BLUR_CONTROLLER_FILTERED_SERVER_CLASS = "blur.controller.filtered.server.class"; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-util/src/main/java/org/apache/blur/zookeeper/ZookeeperPathConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/zookeeper/ZookeeperPathConstants.java b/blur-util/src/main/java/org/apache/blur/zookeeper/ZookeeperPathConstants.java new file mode 100644 index 0000000..325ef6e --- /dev/null +++ b/blur-util/src/main/java/org/apache/blur/zookeeper/ZookeeperPathConstants.java @@ -0,0 +1,74 @@ +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. + */ + +public class ZookeeperPathConstants { + + public static String getBasePath() { + return "/blur"; + } + + public static String getClusterPath(String cluster) { + return getClustersPath() + "/" + cluster; + } + + public static String getClustersPath() { + return getBasePath() + "/clusters"; + } + + public static String getOnlineControllersPath() { + return getBasePath() + "/online-controller-nodes"; + } + + public static String getControllersPath() { + return getBasePath() + "/controller-nodes"; + } + + public static String getOnlineShardsPath(String cluster) { + return getClusterPath(cluster) + "/online-nodes"; + } + + public static String getTablesPath(String cluster) { + return getClusterPath(cluster) + "/tables"; + } + + public static String getTablePath(String cluster, String table) { + return getTablesPath(cluster) + "/" + table; + } + + public static String getSafemodePath(String cluster) { + return getClusterPath(cluster) + "/safemode"; + } + + public static String getShutdownPath(String cluster) { + return getClusterPath(cluster) + "/shutdown"; + } + + public static String getRegisteredShardsPath(String cluster) { + return getClusterPath(cluster) + "/registered-nodes"; + } + + public static String getShardLayoutPath(String cluster) { + return getClusterPath(cluster) + "/layout"; + } + + public static String getShardLayoutPathLocks(String cluster) { + return getShardLayoutPath(cluster) + "/locks"; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/64dc0ae2/blur-util/src/main/resources/blur-default.properties ---------------------------------------------------------------------- diff --git a/blur-util/src/main/resources/blur-default.properties b/blur-util/src/main/resources/blur-default.properties index 33cab4e..afaabf5 100644 --- a/blur-util/src/main/resources/blur-default.properties +++ b/blur-util/src/main/resources/blur-default.properties @@ -42,7 +42,7 @@ blur.shard.hostname= # The binding address of the shard blur.shard.bind.address=0.0.0.0 -# The default binding port of the shard server +# The default binding port of the shard server, 0 for random blur.shard.bind.port=40020 # The number of fetcher threads @@ -175,7 +175,7 @@ blur.max.heap.per.row.fetch=10000000 # The maximum number of records in a single row fetch blur.max.records.per.row.fetch.request=1000 -# The http status page port for the shard server +# The http status page port for the shard server, 0 for random blur.gui.shard.port=40090 # To intercept the calls made to the shard server and perform server side changes to the calls extend org.apache.blur.server.FilteredBlurServer. @@ -216,7 +216,7 @@ blur.controller.hostname= # The binding address of the controller blur.controller.bind.address=0.0.0.0 -# The default binding port of the controller server +# The default binding port of the controller server, 0 for random blur.controller.bind.port=40010 # The connection timeout, NOTE: this will be the maximum amount of time you can wait for a query. @@ -267,7 +267,7 @@ blur.controller.retry.max.mutate.delay=2000 # The ending backoff delay for the last retry for a all other request errors blur.controller.retry.max.default.delay=2000 -# The http status page port for the controller server +# The http status page port for the controller server, 0 for random blur.gui.controller.port=40080 # To intercept the calls made to the controller server and perform server side changes to the calls extend org.apache.blur.server.FilteredBlurServer.
