Fixing a bug with table name reuse. The layout cache was not being cleared when a table was removed.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/6b893545 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/6b893545 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/6b893545 Branch: refs/heads/console-v2 Commit: 6b89354578a4fd2cd41d074fecd879dc80d672d7 Parents: b5eb479 Author: Aaron McCurry <[email protected]> Authored: Wed Apr 16 22:27:40 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Wed Apr 16 22:27:40 2014 -0400 ---------------------------------------------------------------------- .../indexserver/DistributedIndexServer.java | 7 ++++ .../indexserver/DistributedLayoutFactory.java | 3 ++ .../MasterBasedDistributedLayoutFactory.java | 18 +++++++++- .../blur/thrift/BlurControllerServer.java | 36 ++++++++++++++------ .../org/apache/blur/thrift/BlurClusterTest.java | 22 +++++++++++- 5 files changed, 74 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b893545/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 8858621..516941e 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 @@ -219,6 +219,12 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { public DistributedLayout readCurrentLayout(String table) { throw new RuntimeException("Not implemented"); } + + @Override + public Map<String, ?> getLayoutCache() { + throw new RuntimeException("Not implemented"); + } + }; } @@ -445,6 +451,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private void cleanup() { clearMapOfOldTables(_layout); + clearMapOfOldTables(_distributedLayoutFactory.getLayoutCache()); boolean closed = false; Map<String, Map<String, BlurIndex>> oldIndexesThatNeedToBeClosed = clearMapOfOldTables(_indexes); for (String table : oldIndexesThatNeedToBeClosed.keySet()) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b893545/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 a1547e5..a6e3de0 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 @@ -17,6 +17,7 @@ package org.apache.blur.manager.indexserver; import java.util.List; +import java.util.Map; public interface DistributedLayoutFactory { @@ -25,4 +26,6 @@ public interface DistributedLayoutFactory { DistributedLayout readCurrentLayout(String table); + Map<String, ?> getLayoutCache(); + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b893545/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 7a44173..47a8fde 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 @@ -81,17 +81,23 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac try { String existingStoragePath = findExistingStoragePath(table); if (existingStoragePath == null) { + LOG.info("Existing storage path NOT FOUND for table [{0}]", table, existingStoragePath); return null; } Stat stat = _zooKeeper.exists(existingStoragePath, false); if (stat != null) { + LOG.info("Existing storage path for table [{0}] is [{1}]", table, existingStoragePath); LOG.info("Existing layout found for table [{0}]", table); byte[] data = _zooKeeper.getData(existingStoragePath, false, stat); if (data != null) { return fromBytes(data); + } else { + return null; } + } else { + LOG.info("Existing storage path NOT FOUND for table [{0}] path [{1}]", table, existingStoragePath); + return null; } - return null; } catch (Exception e) { LOG.error("Unknown error during layout read.", e); throw new RuntimeException(e); @@ -101,6 +107,7 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac @Override public DistributedLayout createDistributedLayout(String table, List<String> shardList, List<String> shardServerList, List<String> offlineShardServers) { + 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)) { @@ -109,6 +116,7 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac _cachedLayoutMap.put(table, newLayout); return newLayout; } else { + LOG.info("Layout for table [{0}] is up to date.", table); return layout; } } @@ -206,6 +214,9 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac } } } + if (path == null) { + return null; + } return tableStoragePath + "/" + path; } @@ -347,5 +358,10 @@ public class MasterBasedDistributedLayoutFactory implements DistributedLayoutFac return false; } } + + @Override + public Map<String,?> getLayoutCache() { + return _cachedLayoutMap; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b893545/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 06f5ea3..a1cec15 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 @@ -190,6 +190,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { private ConcurrentMap<String, WatchChildren> _watchForOnlineShardsPerCluster = new ConcurrentHashMap<String, WatchChildren>(); private Timer _preconnectTimer; private Timer _tableContextWarmupTimer; + private long _tableLayoutTimeoutNanos = TimeUnit.SECONDS.toNanos(30); public void init() throws KeeperException, InterruptedException { setupZookeeper(); @@ -525,7 +526,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { } if (!validResults(results, shardCount, blurQuery)) { BlurClientManager.sleep(_defaultDelay, _maxDefaultDelay, retries, _maxDefaultRetries); - Map<String, String> map = _shardServerLayout.get().get(table); + Map<String, String> map = getTableLayout(table); LOG.info("Current layout for table [{0}] is [{1}]", table, map); continue OUTER; } @@ -874,11 +875,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { public Map<String, String> shardServerLayout(String table) throws BlurException, TException { try { checkTable(table); - Map<String, Map<String, String>> layout = _shardServerLayout.get(); - Map<String, String> tableLayout = layout.get(table); - if (tableLayout == null) { - return new HashMap<String, String>(); - } + Map<String, String> tableLayout = getTableLayout(table); return tableLayout; } catch (Exception e) { LOG.error("Unknown error while trying to get shard server layout [{0}]", e, table); @@ -1079,7 +1076,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { String table = mutation.getTable(); int numberOfShards = getShardCount(table); - Map<String, String> tableLayout = _shardServerLayout.get().get(table); + Map<String, String> tableLayout = getTableLayout(table); if (tableLayout.size() != numberOfShards) { throw new BException("Cannot update data while shard is missing"); } @@ -1111,7 +1108,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { String table = mutation.getTable(); int numberOfShards = getShardCount(table); - Map<String, String> tableLayout = _shardServerLayout.get().get(table); + Map<String, String> tableLayout = getTableLayout(table); if (tableLayout.size() != numberOfShards) { throw new BException("Cannot update data while shard is missing"); } @@ -1134,6 +1131,25 @@ public class BlurControllerServer extends TableAdmin implements Iface { } } + private Map<String, String> getTableLayout(String table) throws BlurException, TException { + final long s = System.nanoTime(); + + // wait for up to limit for value to not be null. + while (true) { + if (s + _tableLayoutTimeoutNanos < System.nanoTime()) { + throw new BException("Could not get table [{0}] layout.", table); + } + Map<String, Map<String, String>> map = _shardServerLayout.get(); + Map<String, String> layout = map.get(table); + if (layout != null) { + return layout; + } else { + String cluster = getCluster(table); + updateLayout(cluster); + } + } + } + private int getShardCount(String table) throws BlurException, TException { Integer numberOfShards = _tableShardCountMap.get(table); if (numberOfShards == null) { @@ -1159,7 +1175,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { String table = mutation.getTable(); int numberOfShards = getShardCount(table); - Map<String, String> tableLayout = _shardServerLayout.get().get(table); + Map<String, String> tableLayout = getTableLayout(table); if (tableLayout == null || tableLayout.size() != numberOfShards) { throw new BException("Cannot update data while shard is missing"); } @@ -1228,7 +1244,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { String table = mutation.getTable(); int numberOfShards = getShardCount(table); - Map<String, String> tableLayout = _shardServerLayout.get().get(table); + Map<String, String> tableLayout = getTableLayout(table); if (tableLayout == null || tableLayout.size() != numberOfShards) { throw new BException("Cannot update data while shard is missing"); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6b893545/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 cbbf548..eb12d5d 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 @@ -325,7 +325,7 @@ public class BlurClusterTest { } -// @Test + // @Test public void testQueryWithSelectorForDeepPagingPerformance() throws BlurException, TException, IOException, InterruptedException { final String tableName = "testQueryWithSelectorForDeepPagingPerformance"; @@ -663,6 +663,26 @@ public class BlurClusterTest { assertEquals(list, terms); } + @Test + public void testTrucateRaceCondition() throws BlurException, TException, IOException, InterruptedException { + String tableName = "testTrucateRaceCondition"; + createTable(tableName); + loadTable(tableName); + List<Connection> connections = BlurClientManager.getConnections(miniCluster.getControllerConnectionStr()); + Iface client1 = BlurClient.getClient(connections.get(0)); + Iface client2 = BlurClient.getClient(connections.get(1)); + TableDescriptor describe = client1.describe(tableName); + client1.disableTable(tableName); + client1.removeTable(tableName, true); + client1.createTable(describe); + + String rowId = UUID.randomUUID().toString(); + RecordMutation mutation = BlurThriftHelper.newRecordMutation("test", rowId, + BlurThriftHelper.newColumn("test", "value"), BlurThriftHelper.newColumn("facetFixed", "test")); + RowMutation rowMutation = BlurThriftHelper.newRowMutation(tableName, rowId, mutation); + client2.mutate(rowMutation); + } + private void assertRowResults(BlurResults results) { for (BlurResult result : results.getResults()) { assertNull(result.locationId);
