Fixed BLUR-201
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/2ff3c312 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/2ff3c312 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/2ff3c312 Branch: refs/heads/master Commit: 2ff3c31248512863cb81e2e83c65f325ea9b5c02 Parents: 4a97135 Author: Aaron McCurry <[email protected]> Authored: Mon Aug 12 13:11:15 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Mon Aug 12 13:11:15 2013 -0400 ---------------------------------------------------------------------- .../clusterstatus/ZookeeperClusterStatus.java | 87 +++++++++++++------- .../blur/thrift/ThriftBlurControllerServer.java | 2 +- .../blur/thrift/ThriftBlurShardServer.java | 2 +- .../org/apache/blur/mapreduce/BlurTask.java | 2 +- .../org/apache/blur/utils/BlurConstants.java | 5 +- .../src/main/resources/blur-site.properties | 10 +++ 6 files changed, 74 insertions(+), 34 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/2ff3c312/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 7fecccf..9208be0 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 @@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import org.apache.blur.BlurConfiguration; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.lucene.search.FairSimilarity; @@ -41,6 +42,7 @@ import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TJSONProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TMemoryInputTransport; import org.apache.blur.thrift.generated.ColumnPreCache; import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.blur.utils.BlurConstants; import org.apache.blur.utils.BlurUtil; import org.apache.blur.zookeeper.WatchChildren; import org.apache.blur.zookeeper.WatchChildren.OnChange; @@ -61,26 +63,28 @@ public class ZookeeperClusterStatus extends ClusterStatus { private static final Log LOG = LogFactory.getLog(ZookeeperClusterStatus.class); - private ZooKeeper _zk; - private AtomicBoolean _running = new AtomicBoolean(); - private ConcurrentMap<String, Long> _safeModeMap = new ConcurrentHashMap<String, Long>(); - private ConcurrentMap<String, List<String>> _onlineShardsNodes = new ConcurrentHashMap<String, List<String>>(); - private ConcurrentMap<String, Set<String>> _tablesPerCluster = new ConcurrentHashMap<String, Set<String>>(); - private AtomicReference<Set<String>> _clusters = new AtomicReference<Set<String>>(new HashSet<String>()); - private ConcurrentMap<String, Boolean> _enabled = new ConcurrentHashMap<String, Boolean>(); - private ConcurrentMap<String, Boolean> _readOnly = new ConcurrentHashMap<String, Boolean>(); - - private WatchChildren _clusterWatcher; - private ConcurrentMap<String, WatchChildren> _onlineShardsNodesWatchers = new ConcurrentHashMap<String, WatchChildren>(); - private ConcurrentMap<String, WatchChildren> _tableWatchers = new ConcurrentHashMap<String, WatchChildren>(); - private Map<String, SafeModeCacheEntry> _clusterToSafeMode = new ConcurrentHashMap<String, ZookeeperClusterStatus.SafeModeCacheEntry>(); - private ConcurrentMap<String, WatchNodeExistance> _enabledWatchNodeExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); - private ConcurrentMap<String, WatchNodeExistance> _readOnlyWatchNodeExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); - - public ZookeeperClusterStatus(ZooKeeper zooKeeper) { + private final ZooKeeper _zk; + private final BlurConfiguration _configuration; + private final AtomicBoolean _running = new AtomicBoolean(); + private final ConcurrentMap<String, Long> _safeModeMap = new ConcurrentHashMap<String, Long>(); + private final ConcurrentMap<String, List<String>> _onlineShardsNodes = new ConcurrentHashMap<String, List<String>>(); + private final ConcurrentMap<String, Set<String>> _tablesPerCluster = new ConcurrentHashMap<String, Set<String>>(); + private final AtomicReference<Set<String>> _clusters = new AtomicReference<Set<String>>(new HashSet<String>()); + private final ConcurrentMap<String, Boolean> _enabled = new ConcurrentHashMap<String, Boolean>(); + private final ConcurrentMap<String, Boolean> _readOnly = new ConcurrentHashMap<String, Boolean>(); + + private final WatchChildren _clusterWatcher; + private final ConcurrentMap<String, WatchChildren> _onlineShardsNodesWatchers = new ConcurrentHashMap<String, WatchChildren>(); + private final ConcurrentMap<String, WatchChildren> _tableWatchers = new ConcurrentHashMap<String, WatchChildren>(); + private final Map<String, SafeModeCacheEntry> _clusterToSafeMode = new ConcurrentHashMap<String, ZookeeperClusterStatus.SafeModeCacheEntry>(); + private final ConcurrentMap<String, WatchNodeExistance> _enabledWatchNodeExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); + private final ConcurrentMap<String, WatchNodeExistance> _readOnlyWatchNodeExistance = new ConcurrentHashMap<String, WatchNodeExistance>(); + + public ZookeeperClusterStatus(ZooKeeper zooKeeper, BlurConfiguration configuration) { _zk = zooKeeper; _running.set(true); - watchForClusters(); + _clusterWatcher = watchForClusters(); + _configuration = configuration; try { Thread.sleep(1000); } catch (InterruptedException e) { @@ -88,6 +92,23 @@ public class ZookeeperClusterStatus extends ClusterStatus { } } + public ZookeeperClusterStatus(String connectionStr, BlurConfiguration configuration) throws IOException { + this(new ZooKeeper(connectionStr, 30000, new Watcher() { + @Override + public void process(WatchedEvent event) { + + } + }), configuration); + } + + public ZookeeperClusterStatus(ZooKeeper zooKeeper) throws IOException { + this(zooKeeper, new BlurConfiguration()); + } + + public ZookeeperClusterStatus(String connectionStr) throws IOException { + this(connectionStr, new BlurConfiguration()); + } + class Clusters extends OnChange { @Override public void action(List<String> clusters) { @@ -201,17 +222,8 @@ public class ZookeeperClusterStatus extends ClusterStatus { } } - private void watchForClusters() { - _clusterWatcher = new WatchChildren(_zk, ZookeeperPathConstants.getClustersPath()).watch(new Clusters()); - } - - public ZookeeperClusterStatus(String connectionStr) throws IOException { - this(new ZooKeeper(connectionStr, 30000, new Watcher() { - @Override - public void process(WatchedEvent event) { - - } - })); + private WatchChildren watchForClusters() { + return new WatchChildren(_zk, ZookeeperPathConstants.getClustersPath()).watch(new Clusters()); } private String getClusterTableKey(String cluster, String table) { @@ -696,10 +708,10 @@ public class ZookeeperClusterStatus extends ClusterStatus { } String table = BlurUtil.nullCheck(tableDescriptor.name, "tableDescriptor.name cannot be null."); String cluster = BlurUtil.nullCheck(tableDescriptor.cluster, "tableDescriptor.cluster cannot be null."); + assignTableUri(tableDescriptor); String uri = BlurUtil.nullCheck(tableDescriptor.tableUri, "tableDescriptor.tableUri cannot be null."); int shardCount = BlurUtil.zeroCheck(tableDescriptor.shardCount, "tableDescriptor.shardCount cannot be less than 1"); - // @TODO check block size Similarity similarity = BlurUtil.getInstance(tableDescriptor.similarityClass, Similarity.class); boolean blockCaching = tableDescriptor.blockCaching; Set<String> blockCachingFileTypes = tableDescriptor.blockCachingFileTypes; @@ -740,6 +752,21 @@ public class ZookeeperClusterStatus extends ClusterStatus { } } + private void assignTableUri(TableDescriptor tableDescriptor) { + if (tableDescriptor.getTableUri() != null) { + return; + } + String cluster = tableDescriptor.getCluster(); + String defaultTableUriPropertyName = BlurConstants.getDefaultTableUriPropertyName(cluster); + String parentPath = _configuration.get(defaultTableUriPropertyName); + if (parentPath == null) { + return; + } + String tableUri = parentPath + "/" + tableDescriptor.getName(); + LOG.info("Setting default table uri for table [{0}] of [{1}]", tableDescriptor.getName(), tableUri); + tableDescriptor.setTableUri(tableUri); + } + @Override public void disableTable(String cluster, String table) { long s = System.nanoTime(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/2ff3c312/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 a6fb587..4d8a094 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 @@ -98,7 +98,7 @@ public class ThriftBlurControllerServer extends ThriftServer { //@TODO this is confusing because controllers are in a cluster by default, but they see all the shards clusters. BlurUtil.setupZookeeper(zooKeeper, BlurConstants.BLUR_CLUSTER); - final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper); + final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration); BlurControllerServer.BlurClient client = new BlurControllerServer.BlurClientRemote(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/2ff3c312/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 13f6038..aeead2f 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 @@ -176,7 +176,7 @@ public class ThriftBlurShardServer extends ThriftServer { BlurUtil.setupZookeeper(zooKeeper, configuration.get(BLUR_CLUSTER_NAME)); - final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper); + final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration); final BlurIndexRefresher refresher = new BlurIndexRefresher(); refresher.init(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/2ff3c312/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurTask.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurTask.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurTask.java index 8542a43..6750ddb 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurTask.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurTask.java @@ -177,7 +177,7 @@ public class BlurTask implements Writable { return job; } - private void checkTable() { + private void checkTable() throws IOException { ZookeeperClusterStatus status = new ZookeeperClusterStatus(_zooKeeper); // check if table exists String cluster = _tableDescriptor.cluster; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/2ff3c312/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 9c434e3..59fde28 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 @@ -21,7 +21,6 @@ import java.util.concurrent.TimeUnit; import org.apache.blur.BlurConfiguration; - public class BlurConstants { public static final String CONTROLLER = "controller"; @@ -108,4 +107,8 @@ public class BlurConstants { throw new RuntimeException("Unknown error parsing configuration.", e); } } + + public static String getDefaultTableUriPropertyName(String cluster) { + return "blur.cluster." + cluster + ".table.uri"; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/2ff3c312/blur-util/src/main/resources/blur-site.properties ---------------------------------------------------------------------- diff --git a/blur-util/src/main/resources/blur-site.properties b/blur-util/src/main/resources/blur-site.properties index 18ae6f1..3064f67 100644 --- a/blur-util/src/main/resources/blur-site.properties +++ b/blur-util/src/main/resources/blur-site.properties @@ -15,3 +15,13 @@ blur.zookeeper.connection=127.0.0.1 blur.cluster.name=default + +# +# Sets the default table location in hdfs. If left null the table uri property in +# the table descriptor will be required for all tables. +# +# blur.cluster.<cluster name here>.table.uri=<hdfs uri> +# +# Example: +# blur.cluster.default.table.uri=hdfs://namenode/blur/tables +#
