Repository: incubator-blur Updated Branches: refs/heads/apache-blur-0.2 eece2e53f -> dffa41120
Fixing issue with ZooKeeper. When a table was removed because the node was being watched for change the node wasn't actually deleted, although the data was removed from the node. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/44bad350 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/44bad350 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/44bad350 Branch: refs/heads/apache-blur-0.2 Commit: 44bad35003b2d62c102daa9f10e742e50e593208 Parents: eece2e5 Author: Aaron McCurry <[email protected]> Authored: Sat Feb 22 17:23:13 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sat Feb 22 17:23:13 2014 -0500 ---------------------------------------------------------------------- .../clusterstatus/ZookeeperClusterStatus.java | 41 ++++++++++++++++---- 1 file changed, 33 insertions(+), 8 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/44bad350/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 6dc13da..df45b73 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 @@ -131,21 +131,23 @@ public class ZookeeperClusterStatus extends ClusterStatus { } class Tables extends OnChange { - private String cluster; + private final String _cluster; + private final String _tablesPath; public Tables(String cluster) { - this.cluster = cluster; + _cluster = cluster; + _tablesPath = ZookeeperPathConstants.getTablesPath(cluster); } @Override public void action(List<String> tables) { runActions(); - Set<String> newSet = new HashSet<String>(tables); - Set<String> oldSet = _tablesPerCluster.put(cluster, newSet); + Set<String> newSet = new HashSet<String>(filterTables(_tablesPath, tables)); + Set<String> oldSet = _tablesPerCluster.put(_cluster, newSet); Set<String> newTables = getNewTables(newSet, oldSet); Set<String> oldTables = getOldTables(newSet, oldSet); for (String table : oldTables) { - final String clusterTableKey = getClusterTableKey(cluster, table); + final String clusterTableKey = getClusterTableKey(_cluster, table); WatchNodeData watch = _enabledWatchNodeExistance.remove(clusterTableKey); if (watch != null) { watch.close(); @@ -153,8 +155,8 @@ public class ZookeeperClusterStatus extends ClusterStatus { _tableDescriptorCache.remove(table); } for (String table : newTables) { - final String clusterTableKey = getClusterTableKey(cluster, table); - WatchNodeData enabledWatcher = new WatchNodeData(_zk, ZookeeperPathConstants.getTablePath(cluster, table)); + final String clusterTableKey = getClusterTableKey(_cluster, table); + WatchNodeData enabledWatcher = new WatchNodeData(_zk, ZookeeperPathConstants.getTablePath(_cluster, table)); enabledWatcher.watch(new WatchNodeData.OnChange() { @Override public void action(byte[] data) { @@ -401,7 +403,8 @@ public class ZookeeperClusterStatus extends ClusterStatus { long s = System.nanoTime(); try { checkIfOpen(); - return _zk.getChildren(ZookeeperPathConstants.getTablesPath(cluster), false); + String tablesPath = ZookeeperPathConstants.getTablesPath(cluster); + return filterTables(tablesPath, _zk.getChildren(tablesPath, false)); } catch (KeeperException e) { throw new RuntimeException(e); } catch (InterruptedException e) { @@ -412,6 +415,28 @@ public class ZookeeperClusterStatus extends ClusterStatus { } } + private List<String> filterTables(String tablesPath, List<String> tables) { + try { + List<String> result = new ArrayList<String>(); + for (String table : tables) { + String path = tablesPath + "/" + table; + Stat stat = _zk.exists(path, false); + if (stat != null) { + byte[] data = _zk.getData(path, false, stat); + if (data == null) { + LOG.info("Empty table node found at [" + path + "]"); + _zk.delete(path, -1); + } else { + result.add(table); + } + } + } + return result; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + public void close() { if (_running.get()) { _running.set(false);
