diff --git oak-lucene/pom.xml oak-lucene/pom.xml
index 156ab84..d280e67 100644
--- oak-lucene/pom.xml
+++ oak-lucene/pom.xml
@@ -377,5 +377,11 @@
metrics-core
test
+
+ org.apache.commons
+ commons-exec
+ 1.3
+ test
+
diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNode.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNode.java
index 3bac1da..15deaec 100644
--- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNode.java
+++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNode.java
@@ -1,236 +1,57 @@
/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.jackrabbit.oak.plugins.index.lucene;
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkState;
+package org.apache.jackrabbit.oak.plugins.index.lucene;
import java.io.IOException;
-import java.util.Collections;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
import javax.annotation.CheckForNull;
-import javax.annotation.Nullable;
-import com.google.common.collect.Iterables;
-import org.apache.jackrabbit.oak.commons.PathUtils;
-import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndex;
-import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndexFactory;
-import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.ReaderRefreshPolicy;
import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReader;
-import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReaderFactory;
import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriter;
-import org.apache.jackrabbit.oak.spi.state.NodeState;
-import org.apache.jackrabbit.oak.commons.benchmark.PerfLogger;
-import org.apache.lucene.index.IndexReader;
-import org.apache.lucene.index.MultiReader;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.suggest.analyzing.AnalyzingInfixSuggester;
import org.apache.lucene.store.Directory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class IndexNode {
- /**
- * Name of the hidden node under which information about the checkpoints
- * seen and indexed by each async indexer is kept.
- */
- static final String ASYNC = ":async";
-
- private static final AtomicInteger INDEX_NODE_COUNTER = new AtomicInteger();
-
- private static final PerfLogger PERF_LOGGER =
- new PerfLogger(LoggerFactory.getLogger(IndexNode.class.getName() + ".perf"));
-
- static IndexNode open(String indexPath, NodeState root, NodeState defnNodeState,
- LuceneIndexReaderFactory readerFactory, @Nullable NRTIndexFactory nrtFactory)
- throws IOException {
- IndexDefinition definition = new IndexDefinition(root, defnNodeState, indexPath);
- List readers = readerFactory.createReaders(definition, defnNodeState, indexPath);
- NRTIndex nrtIndex = nrtFactory != null ? nrtFactory.createIndex(definition) : null;
- if (!readers.isEmpty() || (nrtIndex != null && !hasAsyncIndexerRun(root))){
- return new IndexNode(PathUtils.getName(indexPath), definition, readers, nrtIndex);
- }
- return null;
- }
-
- static boolean hasAsyncIndexerRun(NodeState root) {
- return root.hasChildNode(ASYNC);
- }
-
- private static final Logger log = LoggerFactory.getLogger(IndexNode.class);
-
- private final List readers;
-
- private final String name;
-
- private final IndexDefinition definition;
-
- private final ReadWriteLock lock = new ReentrantReadWriteLock();
-
- private volatile IndexSearcher indexSearcher;
-
- private final NRTIndex nrtIndex;
-
- private final ReaderRefreshPolicy refreshPolicy;
- private final Runnable refreshCallback = new Runnable() {
- @Override
- public void run() {
- refreshReaders();
- }
- };
+public interface IndexNode {
- private boolean closed = false;
+ void release();
- private List nrtReaders;
+ IndexSearcher getSearcher();
- private final int indexNodeId = INDEX_NODE_COUNTER.incrementAndGet();
+ IndexDefinition getDefinition();
- IndexNode(String name, IndexDefinition definition, List readers, @Nullable NRTIndex nrtIndex)
- throws IOException {
- checkArgument(!readers.isEmpty() || nrtIndex != null);
- this.name = name;
- this.definition = definition;
- this.readers = readers;
- this.nrtIndex = nrtIndex;
- this.nrtReaders = getNRTReaders();
- this.indexSearcher = new IndexSearcher(createReader(nrtReaders));
- this.refreshPolicy = nrtIndex != null ? nrtIndex.getRefreshPolicy() : ReaderRefreshPolicy.NEVER;
- }
-
- String getName() {
- return name;
- }
-
- IndexDefinition getDefinition() {
- return definition;
- }
-
- public IndexSearcher getSearcher() {
- return indexSearcher;
- }
-
- @CheckForNull
- Directory getSuggestDirectory() {
- return readers.isEmpty() ? null : getDefaultReader().getSuggestDirectory();
- }
+ List getPrimaryReaders();
@CheckForNull
- AnalyzingInfixSuggester getLookup() {
- return readers.isEmpty() ? null : getDefaultReader().getLookup();
- }
-
- boolean acquire() {
- lock.readLock().lock();
- if (closed) {
- lock.readLock().unlock();
- return false;
- } else {
- boolean success = false;
- try {
- refreshPolicy.refreshOnReadIfRequired(refreshCallback);
- success = true;
- return true;
- } finally {
- if (!success) {
- lock.readLock().unlock();
- }
- }
- }
- }
-
- public void release() {
- lock.readLock().unlock();
- }
-
- public int getIndexNodeId() {
- return indexNodeId;
- }
+ Directory getSuggestDirectory();
- void close() throws IOException {
- lock.writeLock().lock();
- try {
- checkState(!closed);
- closed = true;
- } finally {
- lock.writeLock().unlock();
- }
-
- //Do not close the NRTIndex here as it might be in use
- //by newer IndexNode. Just close the readers obtained from
- //them
- for (LuceneIndexReader reader : Iterables.concat(readers, nrtReaders)){
- reader.close();
- }
- }
-
- List getPrimaryReaders() {
- return readers;
- }
+ List getNRTReaders();
@CheckForNull
- public LuceneIndexWriter getLocalWriter() throws IOException{
- return nrtIndex != null ? nrtIndex.getWriter() : null;
- }
-
- public void refreshReadersOnWriteIfRequired() {
- refreshPolicy.refreshOnWriteIfRequired(refreshCallback);
- }
-
- private void refreshReaders(){
- long start = PERF_LOGGER.start();
- List newNRTReaders = getNRTReaders();
- //The list reference would differ if index got updated
- //so if they are same no need to reinitialize the searcher
- if (newNRTReaders != nrtReaders) {
- nrtReaders = newNRTReaders;
- indexSearcher = new IndexSearcher(createReader(nrtReaders));
- PERF_LOGGER.end(start, 0, "Refreshed reader for index [{}]", definition);
- }
- }
+ AnalyzingInfixSuggester getLookup();
- private LuceneIndexReader getDefaultReader(){
- //TODO This is still required to support Suggester, Spellcheck etc OAK-4643
- return readers.get(0);
- }
-
- private IndexReader createReader(List nrtReaders) {
- if (readers.size() == 1 && nrtReaders.isEmpty()){
- return readers.get(0).getReader();
- }
- if (nrtReaders.size() == 1 && readers.isEmpty()){
- return nrtReaders.get(0).getReader();
- }
- IndexReader[] readerArr = new IndexReader[readers.size() + nrtReaders.size()];
- int i = 0;
- for (LuceneIndexReader r : Iterables.concat(readers, nrtReaders)){
- readerArr[i++] = r.getReader();
- }
- return new MultiReader(readerArr, true);
- }
-
- List getNRTReaders() {
- return nrtIndex != null ? nrtIndex.getReaders() : Collections.emptyList();
- }
+ int getIndexNodeId();
+ @CheckForNull
+ LuceneIndexWriter getLocalWriter() throws IOException;
+ void refreshReadersOnWriteIfRequired();
}
diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeManager.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeManager.java
new file mode 100644
index 0000000..4373682
--- /dev/null
+++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeManager.java
@@ -0,0 +1,343 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+
+import javax.annotation.CheckForNull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.Iterables;
+import org.apache.jackrabbit.oak.commons.PathUtils;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndex;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndexFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.ReaderRefreshPolicy;
+import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReader;
+import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReaderFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriter;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.commons.benchmark.PerfLogger;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.MultiReader;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.suggest.analyzing.AnalyzingInfixSuggester;
+import org.apache.lucene.store.Directory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IndexNodeManager {
+ /**
+ * Name of the hidden node under which information about the checkpoints
+ * seen and indexed by each async indexer is kept.
+ */
+ static final String ASYNC = ":async";
+
+ private static final AtomicInteger INDEX_NODE_COUNTER = new AtomicInteger();
+
+ private static final PerfLogger PERF_LOGGER =
+ new PerfLogger(LoggerFactory.getLogger(IndexNodeManager.class.getName() + ".perf"));
+
+ static IndexNodeManager open(String indexPath, NodeState root, NodeState defnNodeState,
+ LuceneIndexReaderFactory readerFactory, @Nullable NRTIndexFactory nrtFactory)
+ throws IOException {
+ IndexDefinition definition = new IndexDefinition(root, defnNodeState, indexPath);
+ List readers = readerFactory.createReaders(definition, defnNodeState, indexPath);
+ NRTIndex nrtIndex = nrtFactory != null ? nrtFactory.createIndex(definition) : null;
+ if (!readers.isEmpty() || (nrtIndex != null && !hasAsyncIndexerRun(root))){
+ return new IndexNodeManager(PathUtils.getName(indexPath), definition, readers, nrtIndex);
+ }
+ return null;
+ }
+
+ static boolean hasAsyncIndexerRun(NodeState root) {
+ return root.hasChildNode(ASYNC);
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(IndexNodeManager.class);
+
+ private final List readers;
+
+ private final String name;
+
+ private final IndexDefinition definition;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private volatile SearcherHolder searcherHolder;
+
+ private final NRTIndex nrtIndex;
+
+ private final ReaderRefreshPolicy refreshPolicy;
+
+ private final Runnable refreshCallback = new Runnable() {
+ @Override
+ public void run() {
+ refreshReaders();
+ }
+ };
+
+ private boolean closed = false;
+
+ private final int indexNodeId = INDEX_NODE_COUNTER.incrementAndGet();
+
+ IndexNodeManager(String name, IndexDefinition definition, List readers, @Nullable NRTIndex nrtIndex)
+ throws IOException {
+ checkArgument(!readers.isEmpty() || nrtIndex != null);
+ this.name = name;
+ this.definition = definition;
+ this.readers = readers;
+ this.nrtIndex = nrtIndex;
+ this.searcherHolder = createHolder(getNRTReaders());
+ this.refreshPolicy = nrtIndex != null ? nrtIndex.getRefreshPolicy() : ReaderRefreshPolicy.NEVER;
+ }
+
+ String getName() {
+ return name;
+ }
+
+ IndexDefinition getDefinition() {
+ return definition;
+ }
+
+ @CheckForNull
+ private Directory getSuggestDirectory() {
+ return readers.isEmpty() ? null : getDefaultReader().getSuggestDirectory();
+ }
+
+ @CheckForNull
+ private AnalyzingInfixSuggester getLookup() {
+ return readers.isEmpty() ? null : getDefaultReader().getLookup();
+ }
+
+ @CheckForNull
+ IndexNode acquire() {
+ lock.readLock().lock();
+ if (closed) {
+ lock.readLock().unlock();
+ return null;
+ } else {
+ boolean success = false;
+ try {
+ refreshPolicy.refreshOnReadIfRequired(refreshCallback);
+ success = true;
+ return new IndexNodeImpl(searcherHolder);
+ } finally {
+ if (!success) {
+ lock.readLock().unlock();
+ }
+ }
+ }
+ }
+
+ private void release() {
+ lock.readLock().unlock();
+ }
+
+ private int getIndexNodeId() {
+ return indexNodeId;
+ }
+
+ void close() throws IOException {
+ lock.writeLock().lock();
+ try {
+ checkState(!closed);
+ closed = true;
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ releaseHolder(searcherHolder);
+ closeReaders(readers);
+ }
+
+ private List getPrimaryReaders() {
+ return readers;
+ }
+
+ @CheckForNull
+ private LuceneIndexWriter getLocalWriter() throws IOException{
+ return nrtIndex != null ? nrtIndex.getWriter() : null;
+ }
+
+ private void refreshReadersOnWriteIfRequired() {
+ refreshPolicy.refreshOnWriteIfRequired(refreshCallback);
+ }
+
+ private void refreshReaders(){
+ long start = PERF_LOGGER.start();
+ List newNRTReaders = getNRTReaders();
+ //The list reference would differ if index got updated
+ //so if they are same no need to reinitialize the searcher
+ if (newNRTReaders != searcherHolder.nrtReaders) {
+ SearcherHolder old = searcherHolder;
+ searcherHolder = createHolder(newNRTReaders);
+ releaseHolder(old);
+ PERF_LOGGER.end(start, 0, "Refreshed reader for index [{}]", definition);
+ }
+ }
+
+ private LuceneIndexReader getDefaultReader(){
+ //TODO This is still required to support Suggester, Spellcheck etc OAK-4643
+ return readers.get(0);
+ }
+
+ private IndexReader createReader(List nrtReaders) {
+ //Increment count by 1. MultiReader does it for all readers
+ //So no need for an explicit increment for MultiReader
+
+ if (readers.size() == 1 && nrtReaders.isEmpty()){
+ IndexReader reader = readers.get(0).getReader();
+ reader.incRef();
+ return reader;
+ }
+ if (nrtReaders.size() == 1 && readers.isEmpty()){
+ IndexReader reader = nrtReaders.get(0).getReader();
+ reader.incRef();
+ return reader;
+ }
+
+ IndexReader[] readerArr = new IndexReader[readers.size() + nrtReaders.size()];
+ int i = 0;
+ for (LuceneIndexReader r : Iterables.concat(readers, nrtReaders)){
+ readerArr[i++] = r.getReader();
+ }
+ return new MultiReader(readerArr, false);
+ }
+
+ private List getNRTReaders() {
+ return nrtIndex != null ? nrtIndex.getReaders() : Collections.emptyList();
+ }
+
+ private SearcherHolder createHolder(List newNRTReaders) {
+ return new SearcherHolder(new IndexSearcher(createReader(newNRTReaders)), newNRTReaders);
+ }
+
+ private void closeReaders(Iterable readers) {
+ for (LuceneIndexReader r : readers){
+ try {
+ r.close();
+ } catch (IOException e) {
+ log.warn("Error occurred while releasing reader for index [{}]", definition.getIndexPath(), e);
+ }
+ }
+ }
+
+ private void releaseHolder(SearcherHolder holder) {
+ decrementSearcherUsageCount(holder.searcher);
+ }
+
+ private static void incrementSearcherUsageCount(IndexSearcher searcher) {
+ searcher.getIndexReader().incRef();
+ }
+
+ private void decrementSearcherUsageCount(IndexSearcher searcher) {
+ try {
+ //Decrement the count by 1 as we increased it while creating the searcher
+ //in createReader
+ searcher.getIndexReader().decRef();
+ } catch (IOException e) {
+ log.warn("Error occurred while releasing reader for index [{}]", definition.getIndexPath(), e);
+ }
+ }
+
+ private static class SearcherHolder {
+ final IndexSearcher searcher;
+ final List nrtReaders;
+
+ public SearcherHolder(IndexSearcher searcher, List nrtReaders) {
+ this.searcher = searcher;
+ this.nrtReaders = nrtReaders;
+ }
+ }
+
+ private class IndexNodeImpl implements IndexNode {
+ private final SearcherHolder holder;
+ private final AtomicBoolean released = new AtomicBoolean();
+
+ private IndexNodeImpl(SearcherHolder searcherHolder) {
+ this.holder = searcherHolder;
+ //Increment on each acquire
+ incrementSearcherUsageCount(holder.searcher);
+ }
+
+ @Override
+ public void release() {
+ if (released.compareAndSet(false, true)) {
+ //Decrement on each release
+ decrementSearcherUsageCount(holder.searcher);
+ IndexNodeManager.this.release();
+ }
+ }
+
+ @Override
+ public IndexSearcher getSearcher() {
+ return holder.searcher;
+ }
+
+ @Override
+ public IndexDefinition getDefinition() {
+ return definition;
+ }
+
+ @Override
+ public List getPrimaryReaders() {
+ return IndexNodeManager.this.getPrimaryReaders();
+ }
+
+ @Override
+ public Directory getSuggestDirectory() {
+ return IndexNodeManager.this.getSuggestDirectory();
+ }
+
+ @Override
+ public List getNRTReaders() {
+ return holder.nrtReaders;
+ }
+
+ @Override
+ public AnalyzingInfixSuggester getLookup() {
+ return IndexNodeManager.this.getLookup();
+ }
+
+ @Override
+ public int getIndexNodeId() {
+ return IndexNodeManager.this.getIndexNodeId();
+ }
+
+ @Override
+ public LuceneIndexWriter getLocalWriter() throws IOException {
+ return IndexNodeManager.this.getLocalWriter();
+ }
+
+ @Override
+ public void refreshReadersOnWriteIfRequired() {
+ IndexNodeManager.this.refreshReadersOnWriteIfRequired();
+ }
+ }
+}
diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexTracker.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexTracker.java
index 3381e05..d729f62 100644
--- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexTracker.java
+++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexTracker.java
@@ -16,7 +16,7 @@
*/
package org.apache.jackrabbit.oak.plugins.index.lucene;
-import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Predicates.in;
import static com.google.common.base.Predicates.not;
import static com.google.common.base.Predicates.notNull;
@@ -69,7 +69,7 @@ public class IndexTracker {
private NodeState root = EMPTY_NODE;
- private volatile Map indices = emptyMap();
+ private volatile Map indices = emptyMap();
private volatile boolean refresh;
@@ -91,10 +91,10 @@ public class IndexTracker {
}
synchronized void close() {
- Map indices = this.indices;
+ Map indices = this.indices;
this.indices = emptyMap();
- for (Map.Entry entry : indices.entrySet()) {
+ for (Map.Entry entry : indices.entrySet()) {
try {
entry.getValue().close();
} catch (IOException e) {
@@ -115,8 +115,8 @@ public class IndexTracker {
}
private synchronized void diffAndUpdate(final NodeState root) {
- Map original = indices;
- final Map updates = newHashMap();
+ Map original = indices;
+ final Map updates = newHashMap();
Set indexPaths = Sets.newHashSet();
indexPaths.addAll(original.keySet());
@@ -129,7 +129,7 @@ public class IndexTracker {
public void leave(NodeState before, NodeState after) {
try {
long start = PERF_LOGGER.start();
- IndexNode index = IndexNode.open(path, root, after, readerFactory, nrtFactory);
+ IndexNodeManager index = IndexNodeManager.open(path, root, after, readerFactory, nrtFactory);
PERF_LOGGER.end(start, -1, "[{}] Index found to be updated. Reopening the IndexNode", path);
updates.put(path, index); // index can be null
} catch (IOException e) {
@@ -143,7 +143,7 @@ public class IndexTracker {
this.root = root;
if (!updates.isEmpty()) {
- indices = ImmutableMap.builder()
+ indices = ImmutableMap.builder()
.putAll(filterKeys(original, not(in(updates.keySet()))))
.putAll(filterValues(updates, notNull()))
.build();
@@ -155,7 +155,7 @@ public class IndexTracker {
//Given that Tracker is now invoked from a BackgroundObserver
//not a high concern
for (String path : updates.keySet()) {
- IndexNode index = original.get(path);
+ IndexNodeManager index = original.get(path);
try {
if (index != null) {
index.close();
@@ -172,9 +172,10 @@ public class IndexTracker {
}
public IndexNode acquireIndexNode(String path) {
- IndexNode index = indices.get(path);
- if (index != null && index.acquire()) {
- return index;
+ IndexNodeManager index = indices.get(path);
+ IndexNode indexNode = index != null ? index.acquire() : null;
+ if (indexNode != null) {
+ return indexNode;
} else {
return findIndexNode(path);
}
@@ -182,7 +183,7 @@ public class IndexTracker {
@CheckForNull
public IndexDefinition getIndexDefinition(String indexPath){
- IndexNode node = indices.get(indexPath);
+ IndexNodeManager node = indices.get(indexPath);
if (node != null){
//Accessing the definition should not require
//locking as its immutable state
@@ -207,10 +208,10 @@ public class IndexTracker {
// Retry the lookup from acquireIndexNode now that we're
// synchronized. The acquire() call is guaranteed to succeed
// since the close() method is also synchronized.
- IndexNode index = indices.get(path);
+ IndexNodeManager index = indices.get(path);
if (index != null) {
- checkState(index.acquire());
- return index;
+ IndexNode indexNode = index.acquire();
+ return checkNotNull(indexNode);
}
if (badIndexTracker.isIgnoredBadIndex(path)){
@@ -224,15 +225,16 @@ public class IndexTracker {
try {
if (isLuceneIndexNode(node)) {
- index = IndexNode.open(path, root, node, readerFactory, nrtFactory);
+ index = IndexNodeManager.open(path, root, node, readerFactory, nrtFactory);
if (index != null) {
- checkState(index.acquire());
- indices = ImmutableMap.builder()
+ IndexNode indexNode = index.acquire();
+ checkNotNull(indexNode);
+ indices = ImmutableMap.builder()
.putAll(indices)
.put(path, index)
.build();
badIndexTracker.markGoodIndex(path);
- return index;
+ return indexNode;
}
} else if (node.exists()) {
log.warn("Cannot open Lucene Index at path {} as the index is not of type {}", path, TYPE_LUCENE);
diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndex.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndex.java
index 6823174..bdc6ce8 100644
--- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndex.java
+++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndex.java
@@ -22,6 +22,8 @@ package org.apache.jackrabbit.oak.plugins.index.lucene.hybrid;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
@@ -48,8 +50,6 @@ import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.IndexableField;
import org.apache.lucene.search.suggest.analyzing.AnalyzingInfixSuggester;
import org.apache.lucene.store.Directory;
-import org.apache.lucene.store.FSDirectory;
-import org.apache.lucene.store.NRTCachingDirectory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -86,16 +86,22 @@ public class NRTIndex implements Closeable {
private DirectoryReader dirReader;
private boolean closed;
private List readers;
+ private final List openedReaders;
+ private final boolean assertAllReadersClosed;
+
public NRTIndex(IndexDefinition definition, IndexCopier indexCopier,
IndexUpdateListener refreshPolicy, @Nullable NRTIndex previous,
- StatisticsProvider statisticsProvider, NRTDirectoryFactory directoryFactory) {
+ StatisticsProvider statisticsProvider, NRTDirectoryFactory directoryFactory,
+ boolean assertAllReadersClosed) {
this.definition = definition;
this.indexCopier = indexCopier;
this.refreshPolicy = refreshPolicy;
this.previous = previous;
this.statisticsProvider = statisticsProvider;
this.directoryFactory = directoryFactory;
+ this.assertAllReadersClosed = assertAllReadersClosed;
+ this.openedReaders = assertAllReadersClosed ? new LinkedList<>() : Collections.emptyList();
this.refreshTimer = statisticsProvider.getTimer(metricName("REFRESH_TIME"), StatsOptions.METRICS_ONLY);
this.sizeHisto = statisticsProvider.getHistogram(metricName("SIZE"), StatsOptions.METRICS_ONLY);
@@ -104,8 +110,12 @@ public class NRTIndex implements Closeable {
@CheckForNull
LuceneIndexReader getPrimaryReader() {
- DirectoryReader reader = createReader();
- return reader != null ? new NRTReader(reader, directory) : null;
+ DirectoryReader latestReader = createReader();
+ if (latestReader != dirReader) {
+ decrementReaderUseCount(dirReader);
+ dirReader = latestReader;
+ }
+ return latestReader != null ? new NRTReader(latestReader, directory) : null;
}
public LuceneIndexWriter getWriter() throws IOException {
@@ -139,6 +149,9 @@ public class NRTIndex implements Closeable {
if (previousReader != null) {
newReaders.add(previousReader);
}
+
+ decrementReaderUseCount(dirReader);
+
dirReader = latestReader;
readers = ImmutableList.copyOf(newReaders);
return readers;
@@ -148,10 +161,19 @@ public class NRTIndex implements Closeable {
return refreshPolicy;
}
- public synchronized void close() throws IOException {
+ public void close() throws IOException {
if (closed) {
return;
}
+
+ log.debug("[{}] Closing NRTIndex [{}]", definition.getIndexPath(), getName());
+
+ if (dirReader != null){
+ dirReader.close();
+ }
+
+ assertAllReadersAreClosed();
+
if (indexWriter != null) {
//TODO Close call can possibly be speeded up by
//avoiding merge and dropping stuff in memory. To be explored
@@ -180,7 +202,7 @@ public class NRTIndex implements Closeable {
@Override
public String toString() {
- return definition.getIndexPath();
+ return String.format("%s (%s)", definition.getIndexPath(), getName());
}
//For test
@@ -188,6 +210,30 @@ public class NRTIndex implements Closeable {
return indexDir;
}
+ private String getName(){
+ return indexDir != null ? indexDir.getName() : "UNKNOWN";
+ }
+
+ private void assertAllReadersAreClosed() {
+ for (IndexReader r : openedReaders){
+ if (r.getRefCount() != 0){
+ String msg = String.format("Unclosed reader found with refCount %d for index %s", r.getRefCount(), toString());
+ throw new IllegalStateException(msg);
+ }
+ }
+ }
+
+ private void decrementReaderUseCount(IndexReader reader) {
+ try {
+ if (reader != null) {
+ reader.decRef();
+ }
+ } catch (IOException e) {
+ log.warn("[{}] Error occurred while releasing reader instance {}",
+ definition.getIndexPath(), toString(), e);
+ }
+ }
+
/**
* If index was updated then a new reader would be returned otherwise
* existing reader would be returned
@@ -214,6 +260,11 @@ public class NRTIndex implements Closeable {
}
}
ctx.stop();
+
+ if (assertAllReadersClosed && result != null && result != dirReader) {
+ openedReaders.add(result);
+ }
+
return result;
} catch (IOException e) {
log.warn("Error opening index [{}]", e);
@@ -232,6 +283,7 @@ public class NRTIndex implements Closeable {
//config.setRAMBufferSizeMB(1024*1024*25);
indexWriter = new IndexWriter(directory, config);
+ log.debug("[{}] Created NRTIndex [{}]", definition.getIndexPath(), getName());
return new NRTIndexWriter(indexWriter);
}
diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactory.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactory.java
index 87f555b..b18250f 100644
--- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactory.java
+++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactory.java
@@ -57,6 +57,7 @@ public class NRTIndexFactory implements Closeable{
private final long refreshDeltaInSecs;
private final StatisticsProvider statisticsProvider;
private NRTDirectoryFactory directoryFactory = DefaultNRTDirFactory.INSTANCE;
+ private boolean assertAllResourcesClosed = Boolean.getBoolean("oak.lucene.assertAllResourcesClosed");
public NRTIndexFactory(IndexCopier indexCopier, StatisticsProvider statisticsProvider) {
this(indexCopier, Clock.SIMPLE, REFRESH_DELTA_IN_SECS, statisticsProvider);
@@ -80,7 +81,7 @@ public class NRTIndexFactory implements Closeable{
}
String indexPath = definition.getIndexPath();
NRTIndex current = new NRTIndex(definition, indexCopier, getRefreshPolicy(definition),
- getPrevious(indexPath), statisticsProvider, directoryFactory);
+ getPrevious(indexPath), statisticsProvider, directoryFactory, assertAllResourcesClosed);
indexes.put(indexPath, current);
closeLast(indexPath);
return current;
@@ -102,6 +103,14 @@ public class NRTIndexFactory implements Closeable{
this.directoryFactory = directoryFactory;
}
+ /**
+ * Test mode upon which enables assertions to confirm that all readers are closed
+ * by the time NRTIndex is closed
+ */
+ public void setAssertAllResourcesClosed(boolean assertAllResourcesClosed) {
+ this.assertAllResourcesClosed = assertAllResourcesClosed;
+ }
+
private void closeLast(String indexPath) {
List existing = indexes.get(indexPath);
if (existing.size() <= MAX_INDEX_COUNT){
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeManagerTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeManagerTest.java
new file mode 100644
index 0000000..b4c04a8
--- /dev/null
+++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeManagerTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.plugins.index.lucene;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndex;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndexFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.ReaderRefreshPolicy;
+import org.apache.jackrabbit.oak.plugins.index.lucene.reader.DefaultIndexReaderFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReader;
+import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReaderFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.util.IndexDefinitionBuilder;
+import org.apache.jackrabbit.oak.spi.mount.Mounts;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.stats.StatisticsProvider;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.TermQuery;
+import org.apache.lucene.search.TopDocs;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import static com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
+import static org.apache.jackrabbit.oak.plugins.index.lucene.FieldNames.PATH;
+import static org.apache.jackrabbit.oak.plugins.index.lucene.TestUtil.newDoc;
+import static org.apache.jackrabbit.oak.InitialContent.INITIAL_CONTENT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+
+public class IndexNodeManagerTest {
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder(new File("target"));
+
+ private NodeState root = INITIAL_CONTENT;
+ private NodeBuilder builder = root.builder();
+
+ private IndexCopier indexCopier;
+ private NRTIndexFactory nrtFactory;
+ private LuceneIndexReaderFactory readerFactory;
+
+ @Before
+ public void setUp() throws IOException {
+ indexCopier = new IndexCopier(sameThreadExecutor(), temporaryFolder.getRoot());
+ nrtFactory = new NRTIndexFactory(indexCopier, StatisticsProvider.NOOP);
+ readerFactory = new DefaultIndexReaderFactory(Mounts.defaultMountInfoProvider(), indexCopier);
+ LuceneIndexEditorContext.configureUniqueId(builder);
+ }
+
+ @After
+ public void cleanup() throws IOException {
+ nrtFactory.close();
+ indexCopier.close();
+ }
+
+ @Test
+ public void nullIndexNode() throws Exception{
+ assertNull(IndexNodeManager.open("/foo", root, builder.getNodeState(), readerFactory, null));
+ assertNull(IndexNodeManager.open("/foo", root, builder.getNodeState(), readerFactory, nrtFactory));
+ }
+
+ @Test
+ public void nonNullIndex_OnlyNRT() throws Exception{
+ IndexNodeManager nodeManager = IndexNodeManager.open("/foo", root, createNRTIndex(), readerFactory, nrtFactory);
+ IndexNode node = nodeManager.acquire();
+ assertNotNull(node.getSearcher());
+ TopDocs docs = node.getSearcher().search(new TermQuery(new Term(PATH, "/content/en")), 100);
+ assertEquals(0, docs.totalHits);
+ node.release();
+
+ node.getLocalWriter().updateDocument("/content/en", newDoc("/content/en"));
+ node.refreshReadersOnWriteIfRequired();
+
+ node = nodeManager.acquire();
+ docs = node.getSearcher().search(new TermQuery(new Term(PATH, "/content/en")), 100);
+ assertEquals(1, docs.totalHits);
+ }
+
+ @Test
+ public void nullIndex_NonFreshIndex() throws Exception{
+ NodeBuilder builder = createNRTIndex().builder();
+ NodeBuilder rootBuilder = root.builder();
+ rootBuilder.child(IndexNodeManager.ASYNC);
+ assertNull(IndexNodeManager.open("/foo", rootBuilder.getNodeState(), builder.getNodeState(), readerFactory, nrtFactory));
+ }
+
+ @Test
+ public void lockAndRefreshPolicy() throws Exception {
+ NodeState state = createNRTIndex();
+ IndexDefinition definition = new IndexDefinition(root, state, "/foo");
+ NRTIndex nrtIndex = nrtFactory.createIndex(definition);
+ NRTIndex mock = spy(nrtIndex);
+ doReturn(new FailingPolicy()).when(mock).getRefreshPolicy();
+ IndexNodeManager node = new IndexNodeManager("/foo", definition, Collections.emptyList(), mock);
+
+ try {
+ node.acquire();
+ fail();
+ } catch (Exception ignore) {
+
+ }
+
+ node.close();
+ }
+
+ private static NodeState createNRTIndex(){
+ IndexDefinitionBuilder idx = new IndexDefinitionBuilder();
+ idx.indexRule("nt:base").property("foo").propertyIndex();
+ idx.async("async", "sync");
+ return idx.build();
+ }
+
+ private static class FailingPolicy implements ReaderRefreshPolicy {
+
+ @Override
+ public void refreshOnReadIfRequired(Runnable refreshCallback) {
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public void refreshOnWriteIfRequired(Runnable refreshCallback) {
+ throw new IllegalStateException();
+ }
+ }
+}
\ No newline at end of file
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeTest.java
deleted file mode 100644
index eb8efad..0000000
--- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexNodeTest.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.jackrabbit.oak.plugins.index.lucene;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.Collections;
-
-import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndex;
-import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndexFactory;
-import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.ReaderRefreshPolicy;
-import org.apache.jackrabbit.oak.plugins.index.lucene.reader.DefaultIndexReaderFactory;
-import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReader;
-import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReaderFactory;
-import org.apache.jackrabbit.oak.plugins.index.lucene.util.IndexDefinitionBuilder;
-import org.apache.jackrabbit.oak.spi.mount.Mounts;
-import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
-import org.apache.jackrabbit.oak.spi.state.NodeState;
-import org.apache.jackrabbit.oak.stats.StatisticsProvider;
-import org.apache.lucene.index.Term;
-import org.apache.lucene.search.TermQuery;
-import org.apache.lucene.search.TopDocs;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-
-import static com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
-import static org.apache.jackrabbit.oak.plugins.index.lucene.FieldNames.PATH;
-import static org.apache.jackrabbit.oak.plugins.index.lucene.TestUtil.newDoc;
-import static org.apache.jackrabbit.oak.InitialContent.INITIAL_CONTENT;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
-
-public class IndexNodeTest {
- @Rule
- public TemporaryFolder temporaryFolder = new TemporaryFolder(new File("target"));
-
- private NodeState root = INITIAL_CONTENT;
- private NodeBuilder builder = root.builder();
-
- private IndexCopier indexCopier;
- private NRTIndexFactory nrtFactory;
- private LuceneIndexReaderFactory readerFactory;
-
- @Before
- public void setUp() throws IOException {
- indexCopier = new IndexCopier(sameThreadExecutor(), temporaryFolder.getRoot());
- nrtFactory = new NRTIndexFactory(indexCopier, StatisticsProvider.NOOP);
- readerFactory = new DefaultIndexReaderFactory(Mounts.defaultMountInfoProvider(), indexCopier);
- LuceneIndexEditorContext.configureUniqueId(builder);
- }
-
- @After
- public void cleanup() throws IOException {
- nrtFactory.close();
- indexCopier.close();
- }
-
- @Test
- public void nullIndexNode() throws Exception{
- assertNull(IndexNode.open("/foo", root, builder.getNodeState(), readerFactory, null));
- assertNull(IndexNode.open("/foo", root, builder.getNodeState(), readerFactory, nrtFactory));
- }
-
- @Test
- public void nonNullIndex_OnlyNRT() throws Exception{
- IndexNode node = IndexNode.open("/foo", root, createNRTIndex(), readerFactory, nrtFactory);
- assertNotNull(node.getSearcher());
- TopDocs docs = node.getSearcher().search(new TermQuery(new Term(PATH, "/content/en")), 100);
- assertEquals(0, docs.totalHits);
-
- node.getLocalWriter().updateDocument("/content/en", newDoc("/content/en"));
- node.refreshReadersOnWriteIfRequired();
-
- docs = node.getSearcher().search(new TermQuery(new Term(PATH, "/content/en")), 100);
- assertEquals(1, docs.totalHits);
- }
-
- @Test
- public void nullIndex_NonFreshIndex() throws Exception{
- NodeBuilder builder = createNRTIndex().builder();
- NodeBuilder rootBuilder = root.builder();
- rootBuilder.child(IndexNode.ASYNC);
- assertNull(IndexNode.open("/foo", rootBuilder.getNodeState(), builder.getNodeState(), readerFactory, nrtFactory));
- }
-
- @Test
- public void lockAndRefreshPolicy() throws Exception {
- NodeState state = createNRTIndex();
- IndexDefinition definition = new IndexDefinition(root, state, "/foo");
- NRTIndex nrtIndex = nrtFactory.createIndex(definition);
- NRTIndex mock = spy(nrtIndex);
- doReturn(new FailingPolicy()).when(mock).getRefreshPolicy();
- IndexNode node = new IndexNode("/foo", definition, Collections.emptyList(), mock);
-
- try {
- node.acquire();
- fail();
- } catch (Exception ignore) {
-
- }
-
- node.close();
- }
-
- private static NodeState createNRTIndex(){
- IndexDefinitionBuilder idx = new IndexDefinitionBuilder();
- idx.indexRule("nt:base").property("foo").propertyIndex();
- idx.async("async", "sync");
- return idx.build();
- }
-
- private static class FailingPolicy implements ReaderRefreshPolicy {
-
- @Override
- public void refreshOnReadIfRequired(Runnable refreshCallback) {
- throw new IllegalStateException();
- }
-
- @Override
- public void refreshOnWriteIfRequired(Runnable refreshCallback) {
- throw new IllegalStateException();
- }
- }
-}
\ No newline at end of file
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexPlannerTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexPlannerTest.java
index fc66b74..f6266e9 100644
--- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexPlannerTest.java
+++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexPlannerTest.java
@@ -847,11 +847,11 @@ public class IndexPlannerTest {
//------ END - Suggestion/spellcheck plan tests
private IndexNode createIndexNode(IndexDefinition defn, long numOfDocs) throws IOException {
- return new IndexNode("foo", defn, new TestReaderFactory(createSampleDirectory(numOfDocs)).createReaders(defn, EMPTY_NODE, "foo"), null);
+ return new IndexNodeManager("foo", defn, new TestReaderFactory(createSampleDirectory(numOfDocs)).createReaders(defn, EMPTY_NODE, "foo"), null).acquire();
}
private IndexNode createIndexNode(IndexDefinition defn) throws IOException {
- return new IndexNode("foo", defn, new TestReaderFactory(createSampleDirectory()).createReaders(defn, EMPTY_NODE, "foo"), null);
+ return new IndexNodeManager("foo", defn, new TestReaderFactory(createSampleDirectory()).createReaders(defn, EMPTY_NODE, "foo"), null).acquire();
}
private FilterImpl createFilter(String nodeTypeName) {
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/MultiplexingLucenePropertyIndexTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/MultiplexingLucenePropertyIndexTest.java
index 412908c..ccd51e5 100644
--- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/MultiplexingLucenePropertyIndexTest.java
+++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/MultiplexingLucenePropertyIndexTest.java
@@ -140,12 +140,12 @@ public class MultiplexingLucenePropertyIndexTest extends AbstractQueryTest {
LuceneIndexReaderFactory readerFactory = new DefaultIndexReaderFactory(mip, null);
List readers = readerFactory.createReaders(defn, builder.getNodeState(),"/foo");
- IndexNode node = new IndexNode("foo", defn, readers, null);
+ IndexNodeManager node = new IndexNodeManager("foo", defn, readers, null);
//3 Obtain the plan
FilterImpl filter = createFilter("nt:base");
filter.restrictProperty("foo", Operator.EQUAL, PropertyValues.newString("bar"));
- IndexPlanner planner = new IndexPlanner(node, "/foo", filter, Collections.emptyList());
+ IndexPlanner planner = new IndexPlanner(node.acquire(), "/foo", filter, Collections.emptyList());
QueryIndex.IndexPlan plan = planner.getPlan();
//Count should be sum of both readers
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/HybridIndexTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/HybridIndexTest.java
index 06cd989..aabd6fe 100644
--- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/HybridIndexTest.java
+++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/HybridIndexTest.java
@@ -19,6 +19,7 @@
package org.apache.jackrabbit.oak.plugins.index.lucene.hybrid;
+import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@@ -38,6 +39,9 @@ import javax.management.ObjectName;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.PumpStreamHandler;
import org.apache.commons.io.IOUtils;
import org.apache.jackrabbit.JcrConstants;
import org.apache.jackrabbit.oak.Oak;
@@ -85,7 +89,6 @@ import org.apache.lucene.store.NRTCachingDirectory;
import org.apache.lucene.store.NoLockFactory;
import org.apache.lucene.store.SimpleFSDirectory;
import org.junit.After;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -94,10 +97,11 @@ import static com.google.common.collect.ImmutableList.of;
import static com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
import static org.apache.jackrabbit.oak.spi.mount.Mounts.defaultMountInfoProvider;
import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
import static org.junit.Assume.assumeNoException;
public class HybridIndexTest extends AbstractQueryTest {
@@ -107,6 +111,7 @@ public class HybridIndexTest extends AbstractQueryTest {
public TemporaryFolder temporaryFolder = new TemporaryFolder(new File("target"));
private OptionalEditorProvider optionalEditorProvider = new OptionalEditorProvider();
private NRTIndexFactory nrtIndexFactory;
+ private LuceneIndexProvider luceneIndexProvider;
private NodeStore nodeStore;
private DocumentQueue queue;
private Clock clock = new Clock.Virtual();
@@ -115,8 +120,10 @@ public class HybridIndexTest extends AbstractQueryTest {
private long refreshDelta = TimeUnit.SECONDS.toMillis(1);
@After
- public void tearDown(){
+ public void tearDown() throws IOException {
+ luceneIndexProvider.close();
new ExecutorCloser(executorService).close();
+ nrtIndexFactory.close();
}
@Override
@@ -130,9 +137,10 @@ public class HybridIndexTest extends AbstractQueryTest {
MountInfoProvider mip = defaultMountInfoProvider();
nrtIndexFactory = new NRTIndexFactory(copier, clock, TimeUnit.MILLISECONDS.toSeconds(refreshDelta), StatisticsProvider.NOOP);
+ nrtIndexFactory.setAssertAllResourcesClosed(true);
LuceneIndexReaderFactory indexReaderFactory = new DefaultIndexReaderFactory(mip, copier);
IndexTracker tracker = new IndexTracker(indexReaderFactory,nrtIndexFactory);
- LuceneIndexProvider provider = new LuceneIndexProvider(tracker);
+ luceneIndexProvider = new LuceneIndexProvider(tracker);
queue = new DocumentQueue(100, tracker, sameThreadExecutor());
LuceneIndexEditorProvider editorProvider = new LuceneIndexEditorProvider(copier,
tracker,
@@ -147,8 +155,8 @@ public class HybridIndexTest extends AbstractQueryTest {
Oak oak = new Oak(nodeStore)
.with(new InitialContent())
.with(new OpenSecurityProvider())
- .with((QueryIndexProvider) provider)
- .with((Observer) provider)
+ .with((QueryIndexProvider) luceneIndexProvider)
+ .with((Observer) luceneIndexProvider)
.with(localIndexObserver)
.with(editorProvider)
.with(new PropertyIndexEditorProvider())
@@ -341,7 +349,6 @@ public class HybridIndexTest extends AbstractQueryTest {
assertQuery(query, of("/b", "/c"));
}
- @Ignore("OAK-6500")
@Test
public void noFileLeaks() throws Exception{
nrtIndexFactory.setDirectoryFactory(new NRTDirectoryFactory() {
@@ -370,15 +377,16 @@ public class HybridIndexTest extends AbstractQueryTest {
long fileCount5 = createTestDataAndRunAsync("/content/e", 1);
System.out.printf("Open file count - At end %d", getOpenFileCount());
- assertThat(fileCount4, lessThan(fileCount3));
+ assertThat(fileCount4, lessThanOrEqualTo(fileCount3));
}
private long createTestDataAndRunAsync(String parentPath, int count) throws Exception {
createTestData(parentPath, count);
System.out.printf("Open file count - Post creation of %d nodes at %s is %d%n",count, parentPath, getOpenFileCount());
runAsyncIndex();
- System.out.printf("Open file count - Post async run at %s is %d%n",parentPath, getOpenFileCount());
- return getOpenFileCount();
+ long openCount = getOpenFileCount();
+ System.out.printf("Open file count - Post async run at %s is %d%n",parentPath, openCount);
+ return openCount;
}
private void createTestData(String parentPath, int count) throws CommitFailedException {
@@ -393,7 +401,7 @@ public class HybridIndexTest extends AbstractQueryTest {
}
}
- private long getOpenFileCount() throws Exception {
+ private static long getOpenFileCount() throws Exception {
MBeanServer server = ManagementFactory.getPlatformMBeanServer();
ObjectName name = new ObjectName("java.lang:type=OperatingSystem");
Long val = null;
@@ -404,16 +412,30 @@ public class HybridIndexTest extends AbstractQueryTest {
//is the mbean in use. If running on windows the test would be assumed to be true
assumeNoException(e);
}
+ //dumpOpenFilePaths();
return val;
}
+ private static void dumpOpenFilePaths() throws IOException {
+ String jvmName = ManagementFactory.getRuntimeMXBean().getName();
+ String pid = jvmName.split("@")[0];
+
+ CommandLine cl = new CommandLine("/bin/sh");
+ cl.addArguments(new String[]{"-c", "lsof -p "+pid+" | grep '/nrt'"}, false);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DefaultExecutor executor = new DefaultExecutor();
+ executor.setStreamHandler(new PumpStreamHandler(baos));
+ executor.execute(cl);
+ System.out.println(new String(baos.toByteArray()));
+ }
+
private String explain(String query){
String explain = "explain " + query;
return executeQuery(explain, "JCR-SQL2").get(0);
}
private void runAsyncIndex() {
- Runnable async = WhiteboardUtils.getService(wb, Runnable.class, new Predicate() {
+ AsyncIndexUpdate async = (AsyncIndexUpdate) WhiteboardUtils.getService(wb, Runnable.class, new Predicate() {
@Override
public boolean apply(@Nullable Runnable input) {
return input instanceof AsyncIndexUpdate;
@@ -421,6 +443,9 @@ public class HybridIndexTest extends AbstractQueryTest {
});
assertNotNull(async);
async.run();
+ if (async.isFailing()) {
+ fail("AsyncIndexUpdate failed");
+ }
root.refresh();
}
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactoryTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactoryTest.java
index e93a58a..f4d13c2 100644
--- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactoryTest.java
+++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexFactoryTest.java
@@ -56,6 +56,7 @@ public class NRTIndexFactoryTest {
public void setUp() throws IOException {
indexCopier = new IndexCopier(sameThreadExecutor(), temporaryFolder.getRoot());
indexFactory = new NRTIndexFactory(indexCopier, StatisticsProvider.NOOP);
+ indexFactory.setAssertAllResourcesClosed(true);
}
@Test
diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexTest.java
index a278aee..643107b 100644
--- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexTest.java
+++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/hybrid/NRTIndexTest.java
@@ -66,6 +66,7 @@ public class NRTIndexTest {
public void setUp() throws IOException {
indexCopier = new IndexCopier(sameThreadExecutor(), temporaryFolder.getRoot());
indexFactory = new NRTIndexFactory(indexCopier, StatisticsProvider.NOOP);
+ indexFactory.setAssertAllResourcesClosed(true);
LuceneIndexEditorContext.configureUniqueId(builder);
}
@@ -104,6 +105,7 @@ public class NRTIndexTest {
LuceneIndexWriter writer2 = idx.getWriter();
assertSame(writer, writer2);
+ close(readers);
}
@Test
@@ -143,13 +145,18 @@ public class NRTIndexTest {
document.add(newPathField("/a/b"));
writer.updateDocument("/a/b", document);
- assertEquals(1, idx.getPrimaryReader().getReader().numDocs());
+ LuceneIndexReader reader0 = idx.getPrimaryReader();
+ assertEquals(1, reader0.getReader().numDocs());
+ reader0.close();
writer.updateDocument("/a/b", document);
//Update for same path should not lead to deletion
- assertEquals(2, idx.getPrimaryReader().getReader().numDocs());
- assertEquals(0, idx.getPrimaryReader().getReader().numDeletedDocs());
+ LuceneIndexReader reader = idx.getPrimaryReader();
+ assertEquals(2, reader.getReader().numDocs());
+ assertEquals(0, reader.getReader().numDeletedDocs());
+
+ reader.close();
}
@Test
@@ -163,12 +170,16 @@ public class NRTIndexTest {
w1.updateDocument("/a/b", d1);
NRTIndex idx2 = indexFactory.createIndex(idxDefn);
- assertEquals(1, idx2.getReaders().size());
+ List readers = idx2.getReaders();
+ assertEquals(1, readers.size());
+ close(readers);
LuceneIndexWriter w2 = idx2.getWriter();
- assertEquals(2, idx2.getReaders().size());
+ readers = idx2.getReaders();
+ assertEquals(2, readers.size());
assertNotEquals(idx1.getIndexDir(), idx2.getIndexDir());
+ close(readers);
}
@Test
@@ -189,6 +200,15 @@ public class NRTIndexTest {
w1.updateDocument("/a/b", d1);
List readers3 = idx1.getReaders();
assertNotSame(readers2, readers3);
+
+ close(readers2);
+ close(readers3);
+ }
+
+ private void close(List readers) throws IOException {
+ for (LuceneIndexReader r : readers) {
+ r.close();
+ }
}
private IndexDefinition getSyncIndexDefinition(String indexPath) {