Repository: incubator-blur Updated Branches: refs/heads/master 6557950ca -> 2f3c81030
Cleaned up the way FileSystem objects are created. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/c2dcfbf5 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/c2dcfbf5 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/c2dcfbf5 Branch: refs/heads/master Commit: c2dcfbf5e02e0b99d577169c92148c8b49d9071e Parents: 6557950 Author: Aaron McCurry <[email protected]> Authored: Tue Oct 28 08:20:43 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Tue Oct 28 08:20:43 2014 -0400 ---------------------------------------------------------------------- .../clusterstatus/ZookeeperClusterStatus.java | 18 ++++++++------ .../AbstractDistributedIndexServer.java | 4 +-- .../blur/manager/writer/IndexImporter.java | 26 +++++++++----------- .../blur/thrift/ThriftBlurControllerServer.java | 5 ++-- .../blur/thrift/ThriftBlurShardServer.java | 2 +- .../java/org/apache/blur/utils/BlurUtil.java | 8 +++--- 6 files changed, 33 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c2dcfbf5/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 01d7892..e97e50b 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 @@ -49,6 +49,7 @@ import org.apache.blur.zookeeper.WatchNodeData; import org.apache.blur.zookeeper.ZkUtils; import org.apache.blur.zookeeper.ZooKeeperLockManager; import org.apache.blur.zookeeper.ZookeeperPathConstants; +import org.apache.hadoop.conf.Configuration; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; @@ -73,8 +74,10 @@ public class ZookeeperClusterStatus extends ClusterStatus { private final Map<String, SafeModeCacheEntry> _clusterToSafeMode = new ConcurrentHashMap<String, ZookeeperClusterStatus.SafeModeCacheEntry>(); private final ConcurrentMap<String, WatchNodeData> _enabledWatchNodeExistance = new ConcurrentHashMap<String, WatchNodeData>(); private final Set<Action> _tableStateChange = Collections.newSetFromMap(new ConcurrentHashMap<Action, Boolean>()); + private final Configuration _config; - public ZookeeperClusterStatus(ZooKeeper zooKeeper, BlurConfiguration configuration) { + public ZookeeperClusterStatus(ZooKeeper zooKeeper, BlurConfiguration configuration, Configuration config) { + _config = config; _zk = zooKeeper; _running.set(true); _clusterWatcher = new WatchChildren(_zk, ZookeeperPathConstants.getClustersPath()); @@ -87,21 +90,22 @@ public class ZookeeperClusterStatus extends ClusterStatus { } } - public ZookeeperClusterStatus(String connectionStr, BlurConfiguration configuration) throws IOException { + public ZookeeperClusterStatus(String connectionStr, BlurConfiguration configuration, Configuration config) + throws IOException { this(new ZooKeeper(connectionStr, 30000, new Watcher() { @Override public void process(WatchedEvent event) { } - }), configuration); + }), configuration, config); } public ZookeeperClusterStatus(ZooKeeper zooKeeper) throws IOException { - this(zooKeeper, new BlurConfiguration()); + this(zooKeeper, new BlurConfiguration(), new Configuration()); } public ZookeeperClusterStatus(String connectionStr) throws IOException { - this(connectionStr, new BlurConfiguration()); + this(connectionStr, new BlurConfiguration(), new Configuration()); } class Clusters extends OnChange { @@ -551,7 +555,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { if (_zk.exists(blurTablePath, false) != null) { throw new IOException("Table [" + table + "] already exists."); } - BlurUtil.setupFileSystem(uri, shardCount); + BlurUtil.setupFileSystem(uri, shardCount, _config); byte[] bytes = serializeTableDescriptor(tableDescriptor); BlurUtil.createPath(_zk, blurTablePath, bytes); } catch (IOException e) { @@ -660,7 +664,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { String uri = tableDescriptor.getTableUri(); BlurUtil.removeAll(_zk, blurTablePath); if (deleteIndexFiles) { - BlurUtil.removeIndexFiles(uri); + BlurUtil.removeIndexFiles(uri, _config); } } catch (IOException e) { throw new RuntimeException(e); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c2dcfbf5/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractDistributedIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractDistributedIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractDistributedIndexServer.java index 4b392ca..f50b58e 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractDistributedIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractDistributedIndexServer.java @@ -87,7 +87,7 @@ public abstract class AbstractDistributedIndexServer extends AbstractIndexServer checkTable(table); String tableUri = getTableContext(table).getTablePath().toUri().toString(); Path tablePath = new Path(tableUri); - FileSystem fileSystem = FileSystem.get(tablePath.toUri(), _configuration); + FileSystem fileSystem = tablePath.getFileSystem(_configuration); ContentSummary contentSummary = fileSystem.getContentSummary(tablePath); return contentSummary.getLength(); } @@ -101,7 +101,7 @@ public abstract class AbstractDistributedIndexServer extends AbstractIndexServer TableContext tableContext = getTableContext(table); TableDescriptor descriptor = tableContext.getDescriptor(); Path tablePath = new Path(descriptor.tableUri); - FileSystem fileSystem = FileSystem.get(tablePath.toUri(), _configuration); + FileSystem fileSystem = tablePath.getFileSystem(_configuration); if (!fileSystem.exists(tablePath)) { LOG.error("Table [{0}] is missing, defined location [{1}]", table, tablePath.toUri()); throw new RuntimeException("Table [" + table + "] is missing, defined location [" + tablePath.toUri() + "]"); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c2dcfbf5/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java index 838be4d..0d1d19e 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java @@ -28,7 +28,6 @@ import java.util.TimerTask; import java.util.TreeMap; import java.util.TreeSet; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -66,25 +65,31 @@ public class IndexImporter extends TimerTask implements Closeable { private static final String INUSE = ".inuse"; private static final String BADINDEX = ".badindex"; private static final Lock _globalLock = new ReentrantReadWriteLock().writeLock(); + private static final Timer _timer; + + static { + _timer = new Timer("IndexImporter", true); + } + + public static void closeIndexImporterTimer() { + _timer.cancel(); + _timer.purge(); + } private final static Log LOG = LogFactory.getLog(IndexImporter.class); private final BlurIndex _blurIndex; private final ShardContext _shardContext; - private final Timer _timer; private final String _table; private final String _shard; - private final AtomicBoolean _running = new AtomicBoolean(); private final long _cleanupDelay; private long _lastCleanup; public IndexImporter(BlurIndex blurIndex, ShardContext shardContext, TimeUnit refreshUnit, long refreshAmount) { - _running.set(true); _blurIndex = blurIndex; _shardContext = shardContext; - _timer = new Timer("IndexImporter [" + shardContext.getShard() + "/" + shardContext.getTableContext().getTable() - + "]", true); + long period = refreshUnit.toMillis(refreshAmount); _timer.schedule(this, period, period); _table = _shardContext.getTableContext().getTable(); @@ -94,11 +99,7 @@ public class IndexImporter extends TimerTask implements Closeable { @Override public void close() throws IOException { - if (_running.get()) { - _running.set(false); - _timer.cancel(); - _timer.purge(); - } + } @Override @@ -120,9 +121,6 @@ public class IndexImporter extends TimerTask implements Closeable { FileSystem fileSystem = path.getFileSystem(configuration); SortedSet<FileStatus> listStatus; while (true) { - if (!_running.get()) { - return; - } try { listStatus = sort(fileSystem.listStatus(path, new PathFilter() { @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c2dcfbf5/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 8320847..a07fdf9 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 @@ -128,9 +128,11 @@ public class ThriftBlurControllerServer extends ThriftServer { final ZooKeeper zooKeeper = ZkUtils.newZooKeeper(zkConnectionStr, sessionTimeout); + Configuration config = new Configuration(); + BlurUtil.setupZookeeper(zooKeeper, null); - final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration); + final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration, config); int timeout = configuration.getInt(BLUR_CONTROLLER_SHARD_CONNECTION_TIMEOUT, 60000); BlurControllerServer.BlurClient client = new BlurControllerServer.BlurClientRemote(timeout); @@ -148,7 +150,6 @@ public class ThriftBlurControllerServer extends ThriftServer { int numberOfControllerDriverCommandThreads = configuration.getInt(BLUR_CONTROLLER_COMMAND_DRIVER_THREADS, 16); String commandPath = configuration.get(BLUR_COMMAND_LIB_PATH, getCommandLibPath()); - Configuration config = new Configuration(); final ControllerCommandManager controllerCommandManager = new ControllerCommandManager(tmpPath, commandPath, numberOfControllerWorkerCommandThreads, numberOfControllerDriverCommandThreads, Connection.DEFAULT_TIMEOUT, config); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c2dcfbf5/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 eba2cb3..7d0b0ef 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 @@ -189,7 +189,7 @@ public class ThriftBlurShardServer extends ThriftServer { String cluster = configuration.get(BLUR_CLUSTER_NAME, BLUR_CLUSTER); BlurUtil.setupZookeeper(zooKeeper, cluster); - final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration); + final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration, config); final BlurIndexRefresher refresher = new BlurIndexRefresher(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c2dcfbf5/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java index 3da813f..315b841 100644 --- a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java +++ b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java @@ -712,9 +712,9 @@ public class BlurUtil { zookeeper.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } - public static void setupFileSystem(String uri, int shardCount) throws IOException { + public static void setupFileSystem(String uri, int shardCount, Configuration configuration) throws IOException { Path tablePath = new Path(uri); - FileSystem fileSystem = FileSystem.get(tablePath.toUri(), new Configuration()); + FileSystem fileSystem = tablePath.getFileSystem(configuration); if (createPath(fileSystem, tablePath)) { LOG.info("Table uri existed."); validateShardCount(shardCount, fileSystem, tablePath); @@ -844,9 +844,9 @@ public class BlurUtil { zooKeeper.delete(path, -1); } - public static void removeIndexFiles(String uri) throws IOException { + public static void removeIndexFiles(String uri, Configuration configuration) throws IOException { Path tablePath = new Path(uri); - FileSystem fileSystem = FileSystem.get(tablePath.toUri(), new Configuration()); + FileSystem fileSystem = tablePath.getFileSystem(configuration); fileSystem.delete(tablePath, true); }
