Repository: incubator-blur Updated Branches: refs/heads/apache-blur-0.2 c47af820e -> 02379c667
Fixing an issue in the controller with the new location in ZooKeeper for the table layouts. Also updating the blurclustertest to be easier to debug. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/02379c66 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/02379c66 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/02379c66 Branch: refs/heads/apache-blur-0.2 Commit: 02379c6676082baefb35f5ccca144d844ec6961f Parents: c47af82 Author: Aaron McCurry <[email protected]> Authored: Mon Feb 17 21:46:35 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Mon Feb 17 21:46:35 2014 -0500 ---------------------------------------------------------------------- .../clusterstatus/ZookeeperPathConstants.java | 4 - .../blur/thrift/BlurControllerServer.java | 53 ++-- .../org/apache/blur/thrift/BlurClusterTest.java | 261 ++++++++++--------- 3 files changed, 175 insertions(+), 143 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/02379c66/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 index b342d3e..a8099c1 100644 --- 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 @@ -67,10 +67,6 @@ public class ZookeeperPathConstants { return getClusterPath(cluster) + "/layout"; } - public static String getShardLayoutPathTableLayout(String cluster) { - return getShardLayoutPath(cluster) + "/table_layout"; - } - public static String getShardLayoutPathLocks(String cluster) { return getShardLayoutPath(cluster) + "/locks"; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/02379c66/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 5766ad6..f90dffb 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 @@ -28,10 +28,10 @@ import java.util.HashSet; import java.util.IdentityHashMap; import java.util.List; import java.util.Map; -import java.util.TimerTask; import java.util.Map.Entry; import java.util.Set; import java.util.Timer; +import java.util.TimerTask; import java.util.TreeSet; import java.util.UUID; import java.util.concurrent.Callable; @@ -100,7 +100,6 @@ import org.apache.blur.utils.ForkJoin.ParallelCall; import org.apache.blur.zookeeper.WatchChildren; import org.apache.blur.zookeeper.WatchChildren.OnChange; import org.apache.blur.zookeeper.WatchNodeExistance; -import org.apache.blur.zookeeper.ZkUtils; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs.Ids; @@ -183,6 +182,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { private ConcurrentMap<String, WatchNodeExistance> _watchForTablesPerClusterExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); private ConcurrentMap<String, WatchNodeExistance> _watchForOnlineShardsPerClusterExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); private ConcurrentMap<String, WatchChildren> _watchForTablesPerCluster = new ConcurrentHashMap<String, WatchChildren>(); + private ConcurrentMap<String, WatchChildren> _watchForTableLayoutChanges = new ConcurrentHashMap<String, WatchChildren>(); private ConcurrentMap<String, WatchChildren> _watchForOnlineShardsPerCluster = new ConcurrentHashMap<String, WatchChildren>(); private Timer _preconnectTimer; @@ -194,9 +194,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { watchForClusterChanges(); List<String> clusterList = _clusterStatus.getClusterList(false); for (String cluster : clusterList) { - watchForLayoutChanges(cluster); + watchForLayoutChangeEvents(cluster); + updateLayout(cluster); } - updateLayout(); startPreconnectTimer(); } @@ -255,7 +255,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { } for (String cluster : children) { try { - watchForLayoutChanges(cluster); + watchForLayoutChangeEvents(cluster); } catch (KeeperException e) { LOG.error("Unknown error", e); throw new RuntimeException(e); @@ -268,15 +268,13 @@ public class BlurControllerServer extends TableAdmin implements Iface { }); } - private void watchForLayoutChanges(final String cluster) throws KeeperException, InterruptedException { + private void watchForLayoutChangeEvents(final String cluster) throws KeeperException, InterruptedException { WatchNodeExistance we1 = new WatchNodeExistance(_zookeeper, ZookeeperPathConstants.getTablesPath(cluster)); - final String shardLayoutPathTableLayoutPath = ZookeeperPathConstants.getShardLayoutPathTableLayout(cluster); - ZkUtils.mkNodesStr(_zookeeper, shardLayoutPathTableLayoutPath); we1.watch(new WatchNodeExistance.OnChange() { @Override public void action(Stat stat) { if (stat != null) { - watch(cluster, shardLayoutPathTableLayoutPath, _watchForTablesPerCluster); + watchTables(cluster, _watchForTablesPerCluster); } } }); @@ -285,13 +283,17 @@ public class BlurControllerServer extends TableAdmin implements Iface { } } - private void watch(final String cluster, String path, ConcurrentMap<String, WatchChildren> map) { + private void watchTables(final String cluster, ConcurrentMap<String, WatchChildren> map) { + String path = ZookeeperPathConstants.getTablesPath(cluster); + if (map.containsKey(cluster)) { + return; + } WatchChildren watchForTableLayoutChanges = new WatchChildren(_zookeeper, path); watchForTableLayoutChanges.watch(new OnChange() { @Override public void action(List<String> children) { LOG.info("Layout change for cluster [{0}].", cluster); - updateLayout(); + updateLayout(cluster); } }); if (map.putIfAbsent(cluster, watchForTableLayoutChanges) != null) { @@ -299,18 +301,15 @@ public class BlurControllerServer extends TableAdmin implements Iface { } } - private synchronized void updateLayout() { + private synchronized void updateLayout(String cluster) { if (!_clusterStatus.isOpen()) { LOG.warn("The cluster status object has been closed."); return; } - List<String> tableList = _clusterStatus.getTableList(false); + List<String> tableList = _clusterStatus.getTableList(false, cluster); HashMap<String, Map<String, String>> newLayout = new HashMap<String, Map<String, String>>(); for (String table : tableList) { - String cluster = _clusterStatus.getCluster(false, table); - if (cluster == null) { - continue; - } + watchTableLayouts(cluster, table, _watchForTableLayoutChanges); DistributedLayoutFactory distributedLayoutFactory = getDistributedLayoutFactory(cluster); DistributedLayout layout = distributedLayoutFactory.readCurrentLayout(table); if (layout != null) { @@ -324,6 +323,25 @@ public class BlurControllerServer extends TableAdmin implements Iface { _shardServerLayout.set(newLayout); } + private void watchTableLayouts(final String cluster, final String table, ConcurrentMap<String, WatchChildren> map) { + String path = ZookeeperPathConstants.getTablePath(cluster, table); + String key = cluster + "|" + table; + if (map.containsKey(key)) { + return; + } + WatchChildren watchForTableLayoutChanges = new WatchChildren(_zookeeper, path); + watchForTableLayoutChanges.watch(new OnChange() { + @Override + public void action(List<String> children) { + LOG.info("Layout change for cluster [{0}] table [{1}].", cluster, table); + updateLayout(cluster); + } + }); + if (map.putIfAbsent(key, watchForTableLayoutChanges) != null) { + watchForTableLayoutChanges.close(); + } + } + private synchronized DistributedLayoutFactory getDistributedLayoutFactory(String cluster) { DistributedLayoutFactory distributedLayoutFactory = _distributedLayoutFactoryMap.get(cluster); if (distributedLayoutFactory == null) { @@ -375,6 +393,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { close(_watchForOnlineShardsPerCluster.values()); close(_watchForOnlineShardsPerClusterExistance.values()); close(_watchForTablesPerCluster.values()); + close(_watchForTableLayoutChanges.values()); close(_watchForTablesPerClusterExistance.values()); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/02379c66/blur-core/src/test/java/org/apache/blur/thrift/BlurClusterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/thrift/BlurClusterTest.java b/blur-core/src/test/java/org/apache/blur/thrift/BlurClusterTest.java index a3a6cf1..7d79eba 100644 --- a/blur-core/src/test/java/org/apache/blur/thrift/BlurClusterTest.java +++ b/blur-core/src/test/java/org/apache/blur/thrift/BlurClusterTest.java @@ -30,7 +30,6 @@ import java.lang.management.ManagementFactory; import java.lang.management.MemoryMXBean; import java.lang.management.MemoryUsage; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Random; @@ -72,6 +71,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.permission.FsAction; import org.apache.hadoop.fs.permission.FsPermission; import org.apache.zookeeper.KeeperException; +import org.junit.After; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -112,108 +112,89 @@ public class BlurClusterTest { miniCluster.shutdownBlurCluster(); } + @After + public void tearDown() throws BlurException, TException { + Iface client = getClient(); + List<String> tableList = client.tableList(); + for (String table : tableList) { + client.disableTable(table); + client.removeTable(table, true); + } + } + private Iface getClient() { return BlurClient.getClient(miniCluster.getControllerConnectionStr()); } @Test - public void runClusterIntegrationTests() throws BlurException, TException, IOException, InterruptedException, - KeeperException { - start("testCreateTable"); - testCreateTable(); - start("testLoadTable"); - testLoadTable(); - start("testBlurQueryWithRowId"); - testBlurQueryWithRowId(); - start("testForEmptySchema"); - testForEmptySchema(); - start("testAdminCalls"); - testAdminCalls(); - start("testQueryWithSelector"); - testQueryWithSelector(); - start("testQueryWithFacets"); - testQueryWithFacets(); - start("testBatchFetch"); - testBatchFetch(); - start("testLoadTable"); - testQueryCancel(); - start("testBackPressureViaQuery"); - testBackPressureViaQuery(); - start("testTestShardFailover"); - testTestShardFailover(); - start("testTermsList"); - testTermsList(); - start("testCreateDisableAndRemoveTable"); - testCreateDisableAndRemoveTable(); - start("testCreateTableWithCustomType"); - testCreateTableWithCustomType(); - } - - private void testBlurQueryWithRowId() throws BlurException, TException { + public void testBlurQueryWithRowId() throws BlurException, TException, InterruptedException, IOException { + String tableName = "testBlurQueryWithRowId"; + createTable(tableName); + loadTable(tableName); Blur.Iface client = getClient(); BlurQuery blurQuery = new BlurQuery(); Query query = new Query(); query.setQuery("*"); blurQuery.setQuery(query); - BlurResults results1 = client.query("test", blurQuery); + BlurResults results1 = client.query(tableName, blurQuery); assertEquals(numberOfDocs, results1.getTotalResults()); String id1 = results1.getResults().iterator().next().getFetchResult().getRowResult().getRow().getId(); blurQuery.setRowId(id1); query.setRowQuery(false); - BlurResults results2 = client.query("test", blurQuery); + BlurResults results2 = client.query(tableName, blurQuery); assertEquals(1, results2.getTotalResults()); String id2 = results2.getResults().iterator().next().getFetchResult().getRecordResult().getRowid(); assertEquals(id1, id2); + System.out.println("Finished!"); } - private void testAdminCalls() throws BlurException, TException { + @Test + public void testAdminCalls() throws BlurException, TException, IOException, InterruptedException { + String tableName = "testAdminCalls"; + createTable(tableName); + loadTable(tableName); Blur.Iface client = getClient(); List<String> shardClusterList = client.shardClusterList(); assertEquals(1, shardClusterList.size()); assertEquals(BlurConstants.DEFAULT, shardClusterList.get(0)); - Map<String, String> shardServerLayout = client.shardServerLayout("test"); + Map<String, String> shardServerLayout = client.shardServerLayout(tableName); assertEquals(5, shardServerLayout.size()); - Map<String, Map<String, ShardState>> shardServerLayoutState = client.shardServerLayoutState("test"); + Map<String, Map<String, ShardState>> shardServerLayoutState = client.shardServerLayoutState(tableName); assertEquals(5, shardServerLayoutState.size()); List<String> shardServerList = client.shardServerList(BlurConstants.DEFAULT); assertEquals(3, shardServerList.size()); } - private void start(String name) { - System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<"); - System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<"); - System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<"); - System.out.println("Staring [" + name + "]"); - } - - private void testForEmptySchema() throws BlurException, TException, IOException { + @Test + public void testForEmptySchema() throws BlurException, TException, IOException, InterruptedException { + String tableName = "testForEmptySchema"; + createTable(tableName); Blur.Iface client = getClient(); - Schema schema = client.schema("test"); + Schema schema = client.schema(tableName); Map<String, Map<String, ColumnDefinition>> families = schema.getFamilies(); - assertTrue(!families.isEmpty()); - int size = families.size(); - System.out.println(size); + assertTrue(families.isEmpty()); - TableContext tableContext = TableContext.create(client.describe("test")); + TableContext tableContext = TableContext.create(client.describe(tableName)); FieldManager fieldManager = tableContext.getFieldManager(); assertTrue(fieldManager.addColumnDefinition("test-family", "test-column", null, false, "string", false, null)); TableContext.clear(); - Schema newschema = client.schema("test"); + Schema newschema = client.schema(tableName); Map<String, Map<String, ColumnDefinition>> newfamilies = newschema.getFamilies(); assertTrue(!newfamilies.isEmpty()); int newsize = newfamilies.size(); - assertEquals(size + 1, newsize); + assertEquals(1, newsize); } - private void testCreateTableWithCustomType() throws IOException, BlurException, TException { + @Test + public void testCreateTableWithCustomType() throws IOException, BlurException, TException { Blur.Iface client = getClient(); TableDescriptor tableDescriptor = new TableDescriptor(); tableDescriptor.setName("test_type"); @@ -232,18 +213,18 @@ public class BlurClusterTest { assertEquals(TestType.class.getName(), tableProperties.get("blur.fieldtype.customtype1")); } - public void testCreateTable() throws BlurException, TException, IOException { + public void createTable(String tableName) throws BlurException, TException, IOException { Blur.Iface client = getClient(); TableDescriptor tableDescriptor = new TableDescriptor(); - tableDescriptor.setName("test"); + tableDescriptor.setName(tableName); tableDescriptor.setShardCount(5); - tableDescriptor.setTableUri(miniCluster.getFileSystemUri().toString() + "/blur/test"); + tableDescriptor.setTableUri(miniCluster.getFileSystemUri().toString() + "/blur/" + tableName); client.createTable(tableDescriptor); List<String> tableList = client.tableList(); - assertEquals(Arrays.asList("test"), tableList); + assertTrue(tableList.contains(tableName)); } - public void testLoadTable() throws BlurException, TException, InterruptedException { + public void loadTable(String tableName) throws BlurException, TException, InterruptedException { Iface client = getClient(); int maxFacetValue = 100; List<RowMutation> mutations = new ArrayList<RowMutation>(); @@ -253,7 +234,7 @@ public class BlurClusterTest { RecordMutation mutation = BlurThriftHelper.newRecordMutation("test", rowId, BlurThriftHelper.newColumn("test", "value"), BlurThriftHelper.newColumn("facet", Integer.toString(random.nextInt(maxFacetValue)))); - RowMutation rowMutation = BlurThriftHelper.newRowMutation("test", rowId, mutation); + RowMutation rowMutation = BlurThriftHelper.newRowMutation(tableName, rowId, mutation); mutations.add(rowMutation); } long s = System.nanoTime(); @@ -266,7 +247,7 @@ public class BlurClusterTest { blurQueryRow.setQuery(queryRow); blurQueryRow.setUseCacheIfPresent(false); blurQueryRow.setCacheResult(false); - BlurResults resultsRow = client.query("test", blurQueryRow); + BlurResults resultsRow = client.query(tableName, blurQueryRow); assertRowResults(resultsRow); assertEquals(numberOfDocs, resultsRow.getTotalResults()); @@ -275,15 +256,19 @@ public class BlurClusterTest { queryRecord.rowQuery = false; queryRecord.setQuery("test.test:value"); blurQueryRecord.setQuery(queryRecord); - BlurResults resultsRecord = client.query("test", blurQueryRecord); + BlurResults resultsRecord = client.query(tableName, blurQueryRecord); assertRecordResults(resultsRecord); assertEquals(numberOfDocs, resultsRecord.getTotalResults()); - Schema schema = client.schema("test"); + Schema schema = client.schema(tableName); assertFalse(schema.getFamilies().isEmpty()); } - private void testQueryWithSelector() throws BlurException, TException { + @Test + public void testQueryWithSelector() throws BlurException, TException, IOException, InterruptedException { + final String tableName = "testQueryWithSelector"; + createTable(tableName); + loadTable(tableName); Iface client = getClient(); BlurQuery blurQueryRow = new BlurQuery(); Query queryRow = new Query(); @@ -293,7 +278,7 @@ public class BlurClusterTest { blurQueryRow.setCacheResult(false); blurQueryRow.setSelector(new Selector()); - BlurResults resultsRow = client.query("test", blurQueryRow); + BlurResults resultsRow = client.query(tableName, blurQueryRow); // assertRowResults(resultsRow); assertEquals(numberOfDocs, resultsRow.getTotalResults()); @@ -303,7 +288,11 @@ public class BlurClusterTest { } - private void testQueryWithFacets() throws BlurException, TException { + @Test + public void testQueryWithFacets() throws BlurException, TException, IOException, InterruptedException { + final String tableName = "testQueryWithFacets"; + createTable(tableName); + loadTable(tableName); Iface client = getClient(); BlurQuery blurQueryRow = new BlurQuery(); Query queryRow = new Query(); @@ -317,7 +306,7 @@ public class BlurClusterTest { blurQueryRow.addToFacets(new Facet("test.facet:" + i, Long.MAX_VALUE)); } - BlurResults resultsRow = client.query("test", blurQueryRow); + BlurResults resultsRow = client.query(tableName, blurQueryRow); // assertRowResults(resultsRow); assertEquals(numberOfDocs, resultsRow.getTotalResults()); @@ -327,9 +316,13 @@ public class BlurClusterTest { } - public void testBatchFetch() throws BlurException, TException { + @Test + public void testBatchFetch() throws BlurException, TException, InterruptedException, IOException { + String tableName = "testBatchFetch"; + createTable(tableName); + loadTable(tableName); final Iface client = getClient(); - List<String> terms = client.terms("test", null, "rowid", "", (short) 100); + List<String> terms = client.terms(tableName, null, "rowid", "", (short) 100); List<Selector> selectors = new ArrayList<Selector>(); for (String s : terms) { @@ -338,7 +331,7 @@ public class BlurClusterTest { selectors.add(selector); } - List<FetchResult> fetchRowBatch = client.fetchRowBatch("test", selectors); + List<FetchResult> fetchRowBatch = client.fetchRowBatch(tableName, selectors); assertEquals(100, fetchRowBatch.size()); int i = 0; @@ -349,59 +342,75 @@ public class BlurClusterTest { } - public void testQueryCancel() throws BlurException, TException, InterruptedException { - // This will make each collect in the collectors pause 250 ms per collect - // call - IndexManager.DEBUG_RUN_SLOW.set(true); - - final Iface client = getClient(); - final BlurQuery blurQueryRow = new BlurQuery(); - Query queryRow = new Query(); - queryRow.setQuery("test.test:value"); - blurQueryRow.setQuery(queryRow); - blurQueryRow.setUseCacheIfPresent(false); - blurQueryRow.setCacheResult(false); - blurQueryRow.setUuid("1234"); - - final AtomicReference<BlurException> error = new AtomicReference<BlurException>(); - final AtomicBoolean fail = new AtomicBoolean(); - - new Thread(new Runnable() { - @Override - public void run() { - try { - // This call will take several seconds to execute. - client.query("test", blurQueryRow); - fail.set(true); - } catch (BlurException e) { - error.set(e); - } catch (TException e) { - e.printStackTrace(); - fail.set(true); + @Test + public void testQueryCancel() throws BlurException, TException, InterruptedException, IOException { + final String tableName = "testQueryCancel"; + createTable(tableName); + loadTable(tableName); + try { + // This will make each collect in the collectors pause 250 ms per collect + // call + IndexManager.DEBUG_RUN_SLOW.set(true); + + final Iface client = getClient(); + final BlurQuery blurQueryRow = new BlurQuery(); + Query queryRow = new Query(); + queryRow.setQuery("test.test:value"); + blurQueryRow.setQuery(queryRow); + blurQueryRow.setUseCacheIfPresent(false); + blurQueryRow.setCacheResult(false); + blurQueryRow.setUuid("1234"); + + final AtomicReference<BlurException> error = new AtomicReference<BlurException>(); + final AtomicBoolean fail = new AtomicBoolean(); + + new Thread(new Runnable() { + @Override + public void run() { + try { + // This call will take several seconds to execute. + client.query(tableName, blurQueryRow); + fail.set(true); + } catch (BlurException e) { + error.set(e); + } catch (TException e) { + e.printStackTrace(); + fail.set(true); + } } + }).start(); + Thread.sleep(500); + client.cancelQuery(tableName, blurQueryRow.getUuid()); + BlurException blurException = pollForError(error, 10, TimeUnit.SECONDS, null, fail, -1); + if (fail.get()) { + fail("Unknown error, failing test."); } - }).start(); - Thread.sleep(500); - client.cancelQuery("test", blurQueryRow.getUuid()); - BlurException blurException = pollForError(error, 10, TimeUnit.SECONDS, null, fail, -1); - if (fail.get()) { - fail("Unknown error, failing test."); + assertEquals(blurException.getErrorType(), ErrorType.QUERY_CANCEL); + } finally { + IndexManager.DEBUG_RUN_SLOW.set(false); } - assertEquals(blurException.getErrorType(), ErrorType.QUERY_CANCEL); } - public void testBackPressureViaQuery() throws BlurException, TException, InterruptedException { + // @Test + public void testBackPressureViaQuery() throws BlurException, TException, InterruptedException, IOException { // This will make each collect in the collectors pause 250 ms per collect // call - IndexManager.DEBUG_RUN_SLOW.set(true); - runBackPressureViaQuery(); - Thread.sleep(1000); - System.gc(); - System.gc(); - Thread.sleep(1000); + String tableName = "testAdminCalls"; + createTable(tableName); + loadTable(tableName); + try { + IndexManager.DEBUG_RUN_SLOW.set(true); + runBackPressureViaQuery(tableName); + Thread.sleep(1000); + System.gc(); + System.gc(); + Thread.sleep(1000); + } finally { + IndexManager.DEBUG_RUN_SLOW.set(false); + } } - private void runBackPressureViaQuery() throws InterruptedException { + private void runBackPressureViaQuery(final String tableName) throws InterruptedException { final Iface client = getClient(); final BlurQuery blurQueryRow = new BlurQuery(); Query queryRow = new Query(); @@ -435,7 +444,7 @@ public class BlurClusterTest { public void run() { try { // This call will take several seconds to execute. - client.query("test", blurQueryRow); + client.query(tableName, blurQueryRow); fail.set(true); } catch (BlurException e) { System.out.println("-------------------"); @@ -509,16 +518,19 @@ public class BlurClusterTest { return ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax(); } + @Test public void testTestShardFailover() throws BlurException, TException, InterruptedException, IOException, KeeperException { - + final String tableName = "testTestShardFailover"; + createTable(tableName); + loadTable(tableName); Iface client = getClient(); BlurQuery blurQuery = new BlurQuery(); blurQuery.setUseCacheIfPresent(false); Query query = new Query(); query.setQuery("test.test:value"); blurQuery.setQuery(query); - BlurResults results1 = client.query("test", blurQuery); + BlurResults results1 = client.query(tableName, blurQuery); assertEquals(numberOfDocs, results1.getTotalResults()); assertRowResults(results1); @@ -528,15 +540,19 @@ public class BlurClusterTest { Thread.sleep(TimeUnit.SECONDS.toMillis(1)); // This should block until shards have failed over - client.shardServerLayout("test"); + client.shardServerLayout(tableName); - assertEquals(numberOfDocs, client.query("test", blurQuery).getTotalResults()); + assertEquals(numberOfDocs, client.query(tableName, blurQuery).getTotalResults()); } - public void testTermsList() throws BlurException, TException { + @Test + public void testTermsList() throws BlurException, TException, IOException, InterruptedException { + final String tableName = "testTermsList"; + createTable(tableName); + loadTable(tableName); Iface client = getClient(); - List<String> terms = client.terms("test", "test", "test", null, (short) 10); + List<String> terms = client.terms(tableName, "test", "test", null, (short) 10); List<String> list = new ArrayList<String>(); list.add("value"); assertEquals(list, terms); @@ -564,6 +580,7 @@ public class BlurClusterTest { } } + @Test public void testCreateDisableAndRemoveTable() throws IOException, BlurException, TException { Iface client = getClient(); String tableName = UUID.randomUUID().toString();
