Updated Branches: refs/heads/apache-blur-0.2 29e77928a -> 30cc2277f
Refactoring some of the constructors (removing init methods) and making these Closable. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/30cc2277 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/30cc2277 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/30cc2277 Branch: refs/heads/apache-blur-0.2 Commit: 30cc2277f3f6e786cf4decb372499c9dc4546036 Parents: 29e7792 Author: Aaron McCurry <[email protected]> Authored: Tue Oct 15 14:30:47 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Tue Oct 15 14:30:47 2013 -0400 ---------------------------------------------------------------------- .../indexserver/CloseableExecutorService.java | 4 ++ .../indexserver/DistributedIndexServer.java | 73 ++++++++++---------- .../manager/indexserver/LocalIndexServer.java | 11 +-- .../blur/manager/writer/BlurIndexCloser.java | 7 +- .../blur/manager/writer/BlurIndexRefresher.java | 15 ++-- .../blur/thrift/ThriftBlurShardServer.java | 1 - .../manager/writer/BlurIndexReaderTest.java | 6 +- .../blur/manager/writer/BlurNRTIndexTest.java | 2 - .../writer/DirectoryReferenceCounterTest.java | 3 +- .../refcounter/DirectoryReferenceFileGC.java | 4 +- .../store/refcounter/IndexInputCloser.java | 2 +- 11 files changed, 62 insertions(+), 66 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/main/java/org/apache/blur/manager/indexserver/CloseableExecutorService.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/CloseableExecutorService.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/CloseableExecutorService.java index f5900db..3236b3d 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/CloseableExecutorService.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/CloseableExecutorService.java @@ -32,5 +32,9 @@ public class CloseableExecutorService implements Closeable { public void close() throws IOException { executor.shutdownNow(); } + + public static Closeable close(ExecutorService executor) { + return new CloseableExecutorService(executor); + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/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 2fc819c..f0e6590 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 @@ -71,6 +71,8 @@ import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; +import com.google.common.io.Closer; + public class DistributedIndexServer extends AbstractDistributedIndexServer { private static final Log LOG = LogFactory.getLog(DistributedIndexServer.class); @@ -107,7 +109,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private final DirectoryReferenceFileGC _gc; private final WatchChildren _watchOnlineShards; private final SharedMergeScheduler _mergeScheduler; - private final IndexInputCloser _closer; + private final IndexInputCloser _indexInputCloser; private final ExecutorService _searchExecutor; private final BlurIndexRefresher _refresher; private final BlurIndexCloser _indexCloser; @@ -115,6 +117,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { private final ConcurrentMap<String, LayoutEntry> _layout = new ConcurrentHashMap<String, LayoutEntry>(); private final ConcurrentMap<String, Map<String, BlurIndex>> _indexes = new ConcurrentHashMap<String, Map<String, BlurIndex>>(); private final ShardStateManager _shardStateManager = new ShardStateManager(); + private final Closer _closer; public DistributedIndexServer(Configuration configuration, ZooKeeper zookeeper, ClusterStatus clusterStatus, BlurIndexWarmup warmup, BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory, @@ -122,6 +125,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { int shardOpenerThreadCount, int internalSearchThreads, int warmupThreads) throws KeeperException, InterruptedException { super(clusterStatus, configuration, nodeName, cluster); + _closer = Closer.create(); _shardOpenerThreadCount = shardOpenerThreadCount; _zookeeper = zookeeper; _filterCache = filterCache; @@ -131,22 +135,27 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { _warmupThreads = warmupThreads; _blockCacheDirectoryFactory = blockCacheDirectoryFactory; _distributedLayoutFactory = distributedLayoutFactory == null ? getDefaultLayoutFactory() : distributedLayoutFactory; + + _closer.register(_shardStateManager); BlurUtil.setupZookeeper(_zookeeper, _cluster); _openerService = Executors.newThreadPool("shard-opener", _shardOpenerThreadCount); - _gc = new DirectoryReferenceFileGC(); - _gc.init(); - - // @TODO allow for configuration of these - _mergeScheduler = new SharedMergeScheduler(); _searchExecutor = Executors.newThreadPool("internal-search", _internalSearchThreads); _warmupExecutor = Executors.newThreadPool("warmup", _warmupThreads); - _closer = new IndexInputCloser(); - _closer.init(); - _refresher = new BlurIndexRefresher(); - _refresher.init(); - _indexCloser = new BlurIndexCloser(); - _indexCloser.init(); + + _closer.register(CloseableExecutorService.close(_openerService)); + _closer.register(CloseableExecutorService.close(_searchExecutor)); + _closer.register(CloseableExecutorService.close(_warmupExecutor)); + + _gc = _closer.register(new DirectoryReferenceFileGC()); + + // @TODO allow for configuration of these + _mergeScheduler = _closer.register(new SharedMergeScheduler()); + + + _indexInputCloser = _closer.register(new IndexInputCloser()); + _refresher = _closer.register(new BlurIndexRefresher()); + _indexCloser = _closer.register(new BlurIndexCloser()); _timerCacheFlush = setupFlushCacheTimer(); registerMyselfAsMemberOfCluster(); @@ -164,6 +173,20 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { _timerTableWarmer = setupTableWarmer(); _watchOnlineShards = watchForShardServerChanges(); } + + @Override + public void close() throws IOException { + if (_running.get()) { + _running.set(false); + _closer.close(); + closeAllIndexes(); + _timerCacheFlush.purge(); + _timerCacheFlush.cancel(); + + _timerTableWarmer.purge(); + _timerTableWarmer.cancel(); + } + } private DistributedLayoutFactory getDefaultLayoutFactory() { return new DistributedLayoutFactory() { @@ -195,28 +218,6 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } @Override - public void close() { - if (_running.get()) { - _shardStateManager.close(); - _running.set(false); - closeAllIndexes(); - _refresher.close(); - _indexCloser.close(); - _watchOnlineShards.close(); - _timerCacheFlush.purge(); - _timerCacheFlush.cancel(); - - _timerTableWarmer.purge(); - _timerTableWarmer.cancel(); - _closer.close(); - _gc.close(); - _openerService.shutdownNow(); - _searchExecutor.shutdownNow(); - _warmupExecutor.shutdownNow(); - } - } - - @Override public SortedSet<String> getShardListCurrentServerOnly(String table) throws IOException { return new TreeSet<String>(getShardsToServe(table)); } @@ -276,7 +277,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { } } }); - return watchOnlineShards; + return _closer.register(watchOnlineShards); } private Timer setupTableWarmer() { @@ -469,7 +470,7 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { BlurIndexReader reader = new BlurIndexReader(shardContext, dir, _refresher, _indexCloser); index = reader; } else { - BlurNRTIndex writer = new BlurNRTIndex(shardContext, _mergeScheduler, _closer, dir, _gc, _searchExecutor); + BlurNRTIndex writer = new BlurNRTIndex(shardContext, _mergeScheduler, _indexInputCloser, dir, _gc, _searchExecutor); index = writer; } _filterCache.opening(table, shard, index); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java index 7075818..524cfd9 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java @@ -71,15 +71,10 @@ public class LocalIndexServer extends AbstractIndexServer { public LocalIndexServer(TableDescriptor tableDescriptor) throws IOException { _closer = Closer.create(); _tableContext = TableContext.create(tableDescriptor); - _mergeScheduler = new SharedMergeScheduler(); - _indexInputCloser = new IndexInputCloser(); - _indexInputCloser.init(); - _gc = new DirectoryReferenceFileGC(); - _gc.init(); + _mergeScheduler = _closer.register(new SharedMergeScheduler()); + _indexInputCloser = _closer.register(new IndexInputCloser()); + _gc = _closer.register(new DirectoryReferenceFileGC()); _searchExecutor = Executors.newCachedThreadPool(); - _closer.register(_mergeScheduler); - _closer.register(_indexInputCloser); - _closer.register(_gc); _closer.register(new CloseableExecutorService(_searchExecutor)); getIndexes(_tableContext.getTable()); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexCloser.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexCloser.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexCloser.java index e096818..b31838d 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexCloser.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexCloser.java @@ -16,6 +16,7 @@ package org.apache.blur.manager.writer; * See the License for the specific language governing permissions and * limitations under the License. */ +import java.io.Closeable; import java.util.Collection; import java.util.Iterator; import java.util.concurrent.ExecutorService; @@ -29,7 +30,7 @@ import org.apache.blur.log.LogFactory; import org.apache.lucene.index.IndexReader; -public class BlurIndexCloser implements Runnable { +public class BlurIndexCloser implements Runnable, Closeable { private static final Log LOG = LogFactory.getLog(BlurIndexCloser.class); private static final long PAUSE_TIME = TimeUnit.SECONDS.toMillis(1); @@ -37,8 +38,8 @@ public class BlurIndexCloser implements Runnable { private Collection<IndexReader> readers = new LinkedBlockingQueue<IndexReader>(); private AtomicBoolean running = new AtomicBoolean(); private ExecutorService executorService; - - public void init() { + + public BlurIndexCloser() { running.set(true); daemon = new Thread(this); daemon.setDaemon(true); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexRefresher.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexRefresher.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexRefresher.java index cbeaaa7..67181ab 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexRefresher.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexRefresher.java @@ -16,6 +16,7 @@ package org.apache.blur.manager.writer; * See the License for the specific language governing permissions and * limitations under the License. */ +import java.io.Closeable; import java.io.IOException; import java.util.Collection; import java.util.Timer; @@ -27,7 +28,7 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.lucene.store.AlreadyClosedException; -public class BlurIndexRefresher extends TimerTask { +public class BlurIndexRefresher extends TimerTask implements Closeable { private static final Log LOG = LogFactory.getLog(BlurIndexRefresher.class); @@ -35,6 +36,12 @@ public class BlurIndexRefresher extends TimerTask { private long _period = TimeUnit.MINUTES.toMillis(1); private long _delay = _period; private Collection<BlurIndex> _indexes = new LinkedBlockingQueue<BlurIndex>(); + + public BlurIndexRefresher() { + _timer = new Timer("IndexReader-Refresher", true); + _timer.schedule(this, _delay, _period); + LOG.info("Init Complete"); + } public void register(BlurIndex blurIndex) { _indexes.add(blurIndex); @@ -49,12 +56,6 @@ public class BlurIndexRefresher extends TimerTask { _timer.cancel(); } - public void init() { - _timer = new Timer("IndexReader-Refresher", true); - _timer.schedule(this, _delay, _period); - LOG.info("Init Complete"); - } - @Override public void run() { try { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/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 926c9e9..f02bd3e 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 @@ -179,7 +179,6 @@ public class ThriftBlurShardServer extends ThriftServer { final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration); final BlurIndexRefresher refresher = new BlurIndexRefresher(); - refresher.init(); BlurFilterCache filterCache = getFilterCache(configuration); BlurIndexWarmup indexWarmup = getIndexWarmup(configuration); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexReaderTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexReaderTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexReaderTest.java index 59316f3..6e5234d 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexReaderTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexReaderTest.java @@ -72,9 +72,7 @@ public class BlurIndexReaderTest { mergeScheduler = new SharedMergeScheduler(); gc = new DirectoryReferenceFileGC(); - gc.init(); closer = new IndexInputCloser(); - closer.init(); configuration = new Configuration(); service = Executors.newThreadPool("test", 1); @@ -95,8 +93,6 @@ public class BlurIndexReaderTest { ShardContext shardContext = ShardContext.create(tableContext, "test-shard"); refresher = new BlurIndexRefresher(); indexCloser = new BlurIndexCloser(); - refresher.init(); - indexCloser.init(); reader = new BlurIndexReader(shardContext, directory, refresher, indexCloser); } @@ -107,6 +103,8 @@ public class BlurIndexReaderTest { closer.close(); gc.close(); service.shutdownNow(); + refresher.close(); + indexCloser.close(); rm(base); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/test/java/org/apache/blur/manager/writer/BlurNRTIndexTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurNRTIndexTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurNRTIndexTest.java index 21c531e..8010323 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurNRTIndexTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurNRTIndexTest.java @@ -74,9 +74,7 @@ public class BlurNRTIndexTest { mergeScheduler = new SharedMergeScheduler(); gc = new DirectoryReferenceFileGC(); - gc.init(); closer = new IndexInputCloser(); - closer.init(); configuration = new Configuration(); service = Executors.newThreadPool("test", 10); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-core/src/test/java/org/apache/blur/manager/writer/DirectoryReferenceCounterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/DirectoryReferenceCounterTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/DirectoryReferenceCounterTest.java index 7183bed..e077d12 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/DirectoryReferenceCounterTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/DirectoryReferenceCounterTest.java @@ -78,9 +78,7 @@ public class DirectoryReferenceCounterTest { public void testDirectoryReferenceCounter() throws CorruptIndexException, LockObtainFailedException, IOException, InterruptedException { Directory directory = wrap(new RAMDirectory()); DirectoryReferenceFileGC gc = new DirectoryReferenceFileGC(); - gc.init(); IndexInputCloser closer = new IndexInputCloser(); - closer.init(); DirectoryReferenceCounter counter = new DirectoryReferenceCounter(directory, gc, closer); IndexWriterConfig conf = new IndexWriterConfig(LUCENE_VERSION, new KeywordAnalyzer()); IndexWriter writer = new IndexWriter(counter, conf); @@ -117,6 +115,7 @@ public class DirectoryReferenceCounterTest { last.close(); writer.close(); gc.close(); + closer.close(); } private Document getDoc() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java b/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java index 2779024..3e64f66 100644 --- a/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java +++ b/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java @@ -67,8 +67,8 @@ public class DirectoryReferenceFileGC extends TimerTask implements Closeable { } } } - - public void init() { + + public DirectoryReferenceFileGC() { _timer = new Timer("Blur-File-GC", true); _timer.scheduleAtFixedRate(this, _delay, _delay); _queue = new LinkedBlockingQueue<Value>(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/30cc2277/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/IndexInputCloser.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/IndexInputCloser.java b/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/IndexInputCloser.java index 2683f3c..cd4e2d3 100644 --- a/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/IndexInputCloser.java +++ b/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/IndexInputCloser.java @@ -55,7 +55,7 @@ public class IndexInputCloser implements Runnable, Closeable { } } - public void init() { + public IndexInputCloser() { running.set(true); daemon = new Thread(this); daemon.setDaemon(true);
