Repository: incubator-blur Updated Branches: refs/heads/blur-342a 9acdad258 -> 33963185a
removing the offline stuffs and the old layout manager; no more persistent shard nodes. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/33963185 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/33963185 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/33963185 Branch: refs/heads/blur-342a Commit: 33963185a9b7ceb11f9c5a002d3ca3793a1a8fe4 Parents: 9acdad2 Author: Tim <[email protected]> Authored: Tue Jul 29 09:55:39 2014 -0400 Committer: Tim <[email protected]> Committed: Tue Jul 29 09:55:39 2014 -0400 ---------------------------------------------------------------------- .../indexserver/DistributedIndexServer.java | 36 +-- .../indexserver/DistributedLayoutFactory.java | 3 +- .../indexserver/DistributedLayoutManager.java | 246 ------------------- .../MasterBasedDistributedLayoutFactory.java | 10 +- .../DistributedLayoutManagerTest.java | 116 --------- ...MasterBasedDistributedLayoutFactoryTest.java | 44 ++-- 6 files changed, 23 insertions(+), 432 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/33963185/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 351d8fb..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 @@ -58,8 +58,8 @@ import org.apache.blur.thrift.generated.ShardState; import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.utils.BlurUtil; import org.apache.blur.zookeeper.WatchChildren; -import org.apache.blur.zookeeper.ZookeeperPathConstants; 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/33963185/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/33963185/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/33963185/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 e7ad6fa..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 @@ -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/33963185/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/33963185/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) {
