Fixed BLUR-177
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/945f4b30 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/945f4b30 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/945f4b30 Branch: refs/heads/0.2.0-newtypesystem Commit: 945f4b3074f3a0b8a0a61700a91460bb8a7e6d32 Parents: 0c864ec Author: Aaron McCurry <[email protected]> Authored: Thu Aug 1 16:15:46 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Thu Aug 1 16:15:46 2013 -0400 ---------------------------------------------------------------------- .../blur/manager/writer/BlurNRTIndex.java | 2 +- .../manager/writer/SharedMergeScheduler.java | 36 ++++++++++++++++---- 2 files changed, 31 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/945f4b30/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java index c53af2e..b5706ee 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java @@ -89,7 +89,7 @@ public class BlurNRTIndex extends BlurIndex { TieredMergePolicy mergePolicy = (TieredMergePolicy) conf.getMergePolicy(); mergePolicy.setUseCompoundFile(false); - conf.setMergeScheduler(mergeScheduler); + conf.setMergeScheduler(mergeScheduler.getMergeScheduler()); DirectoryReferenceCounter referenceCounter = new DirectoryReferenceCounter(directory, gc, closer); // This directory allows for warm up by adding tracing ability. http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/945f4b30/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java b/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java index f3e51a7..6cd1318 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.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.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; @@ -29,7 +30,7 @@ import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.MergePolicy; import org.apache.lucene.index.MergeScheduler; -public class SharedMergeScheduler extends MergeScheduler implements Runnable { +public class SharedMergeScheduler implements Runnable, Closeable { private static final Log LOG = LogFactory.getLog(SharedMergeScheduler.class); @@ -47,8 +48,7 @@ public class SharedMergeScheduler extends MergeScheduler implements Runnable { } } - @Override - public void merge(IndexWriter writer) throws IOException { + private void mergeIndexWriter(IndexWriter writer) { synchronized (_writers) { if (!_writers.contains(writer)) { LOG.debug("Adding writer to merge [{0}]", writer); @@ -56,6 +56,30 @@ public class SharedMergeScheduler extends MergeScheduler implements Runnable { } } } + + private void removeWriter(IndexWriter writer) { + synchronized (_writers) { + _writers.remove(writer); + } + } + + public MergeScheduler getMergeScheduler() { + return new MergeScheduler() { + + private IndexWriter _writer; + + @Override + public void merge(IndexWriter writer) throws IOException { + _writer = writer; + mergeIndexWriter(writer); + } + + @Override + public void close() throws IOException { + removeWriter(_writer); + } + }; + } @Override public void close() throws IOException { @@ -75,9 +99,9 @@ public class SharedMergeScheduler extends MergeScheduler implements Runnable { synchronized (this) { wait(ONE_SECOND); } - } else if (mergeWriter(writer)) { + } else if (performMergeWriter(writer)) { // there seems to be more merges to do - merge(writer); + mergeIndexWriter(writer); } } catch (InterruptedException e) { LOG.debug("Merging interrupted, exiting."); @@ -88,7 +112,7 @@ public class SharedMergeScheduler extends MergeScheduler implements Runnable { } } - private boolean mergeWriter(IndexWriter writer) throws IOException { + private boolean performMergeWriter(IndexWriter writer) throws IOException { MergePolicy.OneMerge merge = writer.getNextMerge(); if (merge == null) { LOG.debug("No merges to run for [{0}]", writer);
