diff --git a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java index c78f9e6e79..013d37e8ab 100644 --- a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java +++ b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java @@ -716,6 +716,7 @@ public class AsyncIndexUpdate implements Runnable, Closeable { // task will take care of it taskSplitter.maybeSplit(beforeCheckpoint, callback.lease); IndexUpdate indexUpdate = null; + boolean indexingFailed = false; try { NodeBuilder builder = store.getRoot().builder(); @@ -725,7 +726,7 @@ public class AsyncIndexUpdate implements Runnable, Closeable { ImmutableMap.of(IndexConstants.CHECKPOINT_CREATION_TIME, afterTime)); indexUpdate = new IndexUpdate(provider, name, after, builder, callback, callback, info, corruptIndexHandler) - .withMissingProviderStrategy(missingStrategy); + .withMissingProviderStrategy(missingStrategy); configureRateEstimator(indexUpdate); CommitFailedException exception = EditorDiff.process(VisibleEditor.wrap(indexUpdate), before, after); @@ -773,15 +774,18 @@ public class AsyncIndexUpdate implements Runnable, Closeable { callback.lease, name); if (indexUpdate.isReindexingPerformed()) { log.info("[{}] Reindexing completed for indexes: {} in {} ({} ms)", - name, indexUpdate.getReindexStats(), + name, indexUpdate.getReindexStats(), watch, watch.elapsed(TimeUnit.MILLISECONDS)); progressLogged = true; } corruptIndexHandler.markWorkingIndexes(indexUpdate.getUpdatedIndexPaths()); + } catch (Exception e) { + indexingFailed = true; + throw e; } finally { if (indexUpdate != null) { - if (updatePostRunStatus) { + if ( ! indexingFailed) { indexUpdate.commitProgress(IndexCommitCallback.IndexProgress.COMMIT_SUCCEDED); } else { indexUpdate.commitProgress(IndexCommitCallback.IndexProgress.COMMIT_FAILED); diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java index a99812545c..d3f0ef534f 100644 --- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java +++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java @@ -56,6 +56,7 @@ import org.apache.lucene.store.Directory; import org.apache.lucene.store.FSDirectory; import org.apache.lucene.store.FilterDirectory; import org.apache.lucene.store.NoLockFactory; +import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -124,11 +125,17 @@ public class IndexCopier implements CopyOnReadStatsMBean, Closeable { return new CopyOnReadDirectory(this, remote, local, prefetchEnabled, indexPath, executor); } - public Directory wrapForWrite(LuceneIndexDefinition definition, Directory remote, boolean reindexMode, String dirName) throws IOException { - Directory local = createLocalDirForIndexWriter(definition, dirName); + public Directory wrapForWrite(LuceneIndexDefinition definition, Directory remote, + boolean reindexMode, String dirName, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { + Directory local = createLocalDirForIndexWriter(definition, dirName, reindexMode, cowDirecetoryTracker); String indexPath = definition.getIndexPath(); checkIntegrity(indexPath, local, remote); - return new CopyOnWriteDirectory(this, remote, local, reindexMode, indexPath, executor); + + CopyOnWriteDirectory cowDirectory = new CopyOnWriteDirectory(this, remote, local, reindexMode, indexPath, executor); + cowDirecetoryTracker.registerOpenedDirectory(cowDirectory); + + return cowDirectory; } @Override @@ -148,10 +155,16 @@ public class IndexCopier implements CopyOnReadStatsMBean, Closeable { return indexRootDirectory; } - protected Directory createLocalDirForIndexWriter(LuceneIndexDefinition definition, String dirName) throws IOException { + protected Directory createLocalDirForIndexWriter(LuceneIndexDefinition definition, String dirName, + boolean reindexMode, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { String indexPath = definition.getIndexPath(); File indexWriterDir = getIndexDir(definition, indexPath, dirName); + if (reindexMode) { + cowDirecetoryTracker.registerReindexingLocalDirectory(indexWriterDir); + } + //By design indexing in Oak is single threaded so Lucene locking //can be disabled Directory dir = FSDirectory.open(indexWriterDir, NoLockFactory.getNoLockFactory()); @@ -637,4 +650,17 @@ public class IndexCopier implements CopyOnReadStatsMBean, Closeable { } } } + + public interface COWDirecetoryTracker { + void registerOpenedDirectory(@NotNull CopyOnWriteDirectory directory); + void registerReindexingLocalDirectory(@NotNull File dir); + + COWDirecetoryTracker NOOP = new COWDirecetoryTracker() { + @Override + public void registerOpenedDirectory(CopyOnWriteDirectory directory) {} + + @Override + public void registerReindexingLocalDirectory(File dir) {} + }; + } } diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexEditorProvider.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexEditorProvider.java index d0e34cf05c..e5d54bd339 100644 --- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexEditorProvider.java +++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexEditorProvider.java @@ -16,11 +16,15 @@ */ package org.apache.jackrabbit.oak.plugins.index.lucene; +import com.google.common.collect.Lists; +import org.apache.commons.io.FileUtils; import org.apache.jackrabbit.oak.api.CommitFailedException; import org.apache.jackrabbit.oak.plugins.index.*; +import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier.COWDirecetoryTracker; import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory; import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory.ActiveDeletedBlobCollector; import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory.BlobDeletionCallback; +import org.apache.jackrabbit.oak.plugins.index.lucene.directory.CopyOnWriteDirectory; import org.apache.jackrabbit.oak.plugins.index.lucene.directory.DefaultDirectoryFactory; import org.apache.jackrabbit.oak.plugins.index.lucene.directory.DirectoryFactory; import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.IndexingQueue; @@ -50,8 +54,10 @@ import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.File; import java.util.Collection; import java.util.LinkedList; +import java.util.List; import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; @@ -213,7 +219,12 @@ public class LuceneIndexEditorProvider implements IndexEditorProvider { } if (writerFactory == null) { - writerFactory = new DefaultIndexWriterFactory(mountInfoProvider, newDirectoryFactory(blobDeletionCallback), writerConfig); + COWDirectoryCleanupCallback cowDirectoryCleanupCallback = new COWDirectoryCleanupCallback(); + indexingContext.registerIndexCommitCallback(cowDirectoryCleanupCallback); + + writerFactory = new DefaultIndexWriterFactory(mountInfoProvider, + newDirectoryFactory(blobDeletionCallback, cowDirectoryCleanupCallback), + writerConfig); } LuceneIndexEditorContext context = new LuceneIndexEditorContext(root, definition, indexDefinition, callback, @@ -251,8 +262,9 @@ public class LuceneIndexEditorProvider implements IndexEditorProvider { this.inMemoryDocsLimit = inMemoryDocsLimit; } - protected DirectoryFactory newDirectoryFactory(BlobDeletionCallback blobDeletionCallback) { - return new DefaultDirectoryFactory(indexCopier, blobStore, blobDeletionCallback); + protected DirectoryFactory newDirectoryFactory(BlobDeletionCallback blobDeletionCallback, + COWDirecetoryTracker cowDirectoryTracker) { + return new DefaultDirectoryFactory(indexCopier, blobStore, blobDeletionCallback, cowDirectoryTracker); } private LuceneDocumentHolder getDocumentHolder(CommitContext commitContext){ @@ -288,4 +300,46 @@ public class LuceneIndexEditorProvider implements IndexEditorProvider { private static CommitContext getCommitContext(IndexingContext indexingContext) { return (CommitContext) indexingContext.getCommitInfo().getInfo().get(CommitContext.NAME); } + + private static class COWDirectoryCleanupCallback implements IndexCommitCallback, COWDirecetoryTracker { + private static final Logger LOG = LoggerFactory.getLogger(COWDirectoryCleanupCallback.class); + + private List openedCoWDirectories = Lists.newArrayList(); + private List reindexingLocalDirectories = Lists.newArrayList(); + + + @Override + public void commitProgress(IndexProgress indexProgress) { + if (indexProgress != IndexProgress.COMMIT_SUCCEDED && indexProgress != IndexProgress.COMMIT_FAILED) { + LOG.debug("We only care for commit success/failure"); + return; + } + // we only worry about failed indexing + if (indexProgress == IndexProgress.COMMIT_FAILED) { + for (CopyOnWriteDirectory d : openedCoWDirectories) { + try { + d.close(); + } catch (Exception e) { + LOG.warn("Error occurred while closing {}", d, e); + } + } + + for (File f : reindexingLocalDirectories) { + if ( ! FileUtils.deleteQuietly(f)) { + LOG.warn("Failed to delete {}", f); + } + } + } + } + + @Override + public void registerOpenedDirectory(@NotNull CopyOnWriteDirectory directory) { + openedCoWDirectories.add(directory); + } + + @Override + public void registerReindexingLocalDirectory(@NotNull File dir) { + reindexingLocalDirectories.add(dir); + } + } } diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnWriteDirectory.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnWriteDirectory.java index e89a271a05..85129fe140 100644 --- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnWriteDirectory.java +++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnWriteDirectory.java @@ -79,6 +79,7 @@ public class CopyOnWriteDirectory extends FilterDirectory { private final CountDownLatch copyDone = new CountDownLatch(1); private final boolean reindexMode; private final String indexPath; + private boolean closed; /** * Current background task @@ -205,6 +206,10 @@ public class CopyOnWriteDirectory extends FilterDirectory { @Override public void close() throws IOException { + if (closed) { + return; + } + int pendingCopies = queue.size(); addTask(STOP); @@ -257,6 +262,8 @@ public class CopyOnWriteDirectory extends FilterDirectory { local.close(); remote.close(); + + closed = true; } @Override diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/DefaultDirectoryFactory.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/DefaultDirectoryFactory.java index ca2a99d544..2b526cfb06 100644 --- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/DefaultDirectoryFactory.java +++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/DefaultDirectoryFactory.java @@ -22,7 +22,9 @@ package org.apache.jackrabbit.oak.plugins.index.lucene.directory; import java.io.File; import java.io.IOException; +import org.apache.jackrabbit.oak.plugins.index.IndexCommitCallback; import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier; +import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier.COWDirecetoryTracker; import org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexDefinition; import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory.BlobDeletionCallback; import org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants; @@ -41,15 +43,18 @@ public class DefaultDirectoryFactory implements DirectoryFactory { private final IndexCopier indexCopier; private final GarbageCollectableBlobStore blobStore; private final BlobDeletionCallback blobDeletionCallback; + private final COWDirecetoryTracker cowDirecetoryTracker; public DefaultDirectoryFactory(@Nullable IndexCopier indexCopier, @Nullable GarbageCollectableBlobStore blobStore) { - this(indexCopier, blobStore, BlobDeletionCallback.NOOP); + this(indexCopier, blobStore, BlobDeletionCallback.NOOP, COWDirecetoryTracker.NOOP); } public DefaultDirectoryFactory(@Nullable IndexCopier indexCopier, @Nullable GarbageCollectableBlobStore blobStore, - @NotNull ActiveDeletedBlobCollectorFactory.BlobDeletionCallback blobDeletionCallback) { + @NotNull ActiveDeletedBlobCollectorFactory.BlobDeletionCallback blobDeletionCallback, + @NotNull COWDirecetoryTracker cowDirectoryTracker) { this.indexCopier = indexCopier; this.blobStore = blobStore; this.blobDeletionCallback = blobDeletionCallback; + this.cowDirecetoryTracker = cowDirectoryTracker; } @Override @@ -65,7 +70,7 @@ public class DefaultDirectoryFactory implements DirectoryFactory { Directory d = indexCopier.wrapForRead(indexPath, definition, directory, dirName); d.close(); } - directory = indexCopier.wrapForWrite(definition, directory, reindex, dirName); + directory = indexCopier.wrapForWrite(definition, directory, reindex, dirName, cowDirecetoryTracker); } return directory; } diff --git a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierCleanupTest.java b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierCleanupTest.java index 8f47c25bac..2770f2dbcf 100644 --- a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierCleanupTest.java +++ b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierCleanupTest.java @@ -474,7 +474,9 @@ public class IndexCopierCleanupTest { } @Override - protected Directory createLocalDirForIndexWriter(LuceneIndexDefinition definition, String dirName) throws IOException { + protected Directory createLocalDirForIndexWriter(LuceneIndexDefinition definition, String dirName, + boolean reindexMode, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { return new DelayCopyingSimpleFSDirectory(baseFSDir); } @@ -487,7 +489,7 @@ public class IndexCopierCleanupTest { } Directory getCoWDir() throws IOException { - return wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + return wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, COWDirecetoryTracker.NOOP); } } diff --git a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java index f6ecd4ad5e..ac84f107ee 100644 --- a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java +++ b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java @@ -595,7 +595,8 @@ public class IndexCopierTest { IndexCopier copier = new IndexCopier(sameThreadExecutor(), getWorkDir()); LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); - Directory dir = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + Directory dir = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); byte[] t1 = writeFile(dir, "t1"); @@ -613,7 +614,8 @@ public class IndexCopierTest { IndexCopier copier = new IndexCopier(sameThreadExecutor(), getWorkDir()); LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); - Directory dir = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + Directory dir = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); byte[] t1 = writeFile(dir, "t1"); @@ -642,7 +644,8 @@ public class IndexCopierTest { //State of remote directory should set before wrapping as later //additions would not be picked up given COW assume remote directory //to be read only - Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); assertEquals(newHashSet("t1"), newHashSet(local.listAll())); assertEquals(t1.length, local.fileLength("t1")); @@ -699,7 +702,8 @@ public class IndexCopierTest { Directory remote = new CloseSafeDir(); byte[] t1 = writeFile(remote, "t1"); byte[] t2 = writeFile(remote, "t2"); - Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); assertEquals(newHashSet("t1", "t2"), newHashSet(local.listAll())); byte[] t3 = writeFile(local, "t3"); @@ -749,7 +753,8 @@ public class IndexCopierTest { } }; byte[] t1 = writeFile(remote, "t1"); - Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); //Read should be served from remote readRemotes.clear();readLocal.clear(); @@ -779,7 +784,8 @@ public class IndexCopierTest { Directory remote = new CloseSafeDir(); - final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); byte[] t1 = writeFile(local, "t1"); assertTrue(local.fileExists("t1")); @@ -841,7 +847,8 @@ public class IndexCopierTest { Directory remote = new CloseSafeDir(); - final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); byte[] t1 = writeFile(local, "t1"); assertTrue(local.fileExists("t1")); @@ -909,7 +916,8 @@ public class IndexCopierTest { } }; - final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); toFail.add("t2"); byte[] t1 = writeFile(local, "t1"); byte[] t2 = writeFile(local, "t2"); @@ -947,7 +955,8 @@ public class IndexCopierTest { } }; - final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME); + final Directory local = copier.wrapForWrite(defn, remote, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); toPause.add("t2"); byte[] t1 = writeFile(local, "t1"); byte[] t2 = writeFile(local, "t2"); @@ -1018,7 +1027,8 @@ public class IndexCopierTest { }; //Start copying a file to remote via COW - Directory cow1 = copier.wrapForWrite(defn, remote2, false, INDEX_DATA_CHILD_NAME); + Directory cow1 = copier.wrapForWrite(defn, remote2, false, INDEX_DATA_CHILD_NAME, + IndexCopier.COWDirecetoryTracker.NOOP); byte[] f2 = writeFile(cow1, "f2"); //Before copy is done to remote lets delete f1 from remote and @@ -1145,7 +1155,9 @@ public class IndexCopierTest { } @Override - protected Directory createLocalDirForIndexWriter(LuceneIndexDefinition definition, String dirName) throws IOException { + protected Directory createLocalDirForIndexWriter(LuceneIndexDefinition definition, String dirName, + boolean reindexMode, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { return baseDir; } } diff --git a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LucenePropertyIndexTest.java b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LucenePropertyIndexTest.java index bc4c3f919f..c1c258a2df 100644 --- a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LucenePropertyIndexTest.java +++ b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LucenePropertyIndexTest.java @@ -215,8 +215,9 @@ public class LucenePropertyIndexTest extends AbstractQueryTest { @Override public Directory wrapForWrite(LuceneIndexDefinition definition, - Directory remote, boolean reindexMode, String dirName) throws IOException { - Directory ret = super.wrapForWrite(definition, remote, reindexMode, dirName); + Directory remote, boolean reindexMode, String dirName, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { + Directory ret = super.wrapForWrite(definition, remote, reindexMode, dirName, cowDirecetoryTracker); cowDir = getFSDirPath(ret); return ret; } diff --git a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneWritesOnSegmentStatsTest.java b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneWritesOnSegmentStatsTest.java index aa01bdc4eb..cc56632ace 100644 --- a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneWritesOnSegmentStatsTest.java +++ b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneWritesOnSegmentStatsTest.java @@ -235,8 +235,9 @@ public class LuceneWritesOnSegmentStatsTest extends AbstractQueryTest { @Override public Directory wrapForWrite(LuceneIndexDefinition definition, - Directory remote, boolean reindexMode, String dirName) throws IOException { - Directory ret = super.wrapForWrite(definition, remote, reindexMode, dirName); + Directory remote, boolean reindexMode, String dirName, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { + Directory ret = super.wrapForWrite(definition, remote, reindexMode, dirName, cowDirecetoryTracker); cowDir = getFSDirPath(ret); return ret; } diff --git a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/FailedIndexUpdateTest.java b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/FailedIndexUpdateTest.java new file mode 100644 index 0000000000..b1c63f1421 --- /dev/null +++ b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/FailedIndexUpdateTest.java @@ -0,0 +1,383 @@ +/* + * 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.writer; + +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; +import com.google.common.io.Closer; +import org.apache.jackrabbit.oak.Oak; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.api.PropertyState; +import org.apache.jackrabbit.oak.api.Root; +import org.apache.jackrabbit.oak.api.Tree; +import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser; +import org.apache.jackrabbit.oak.plugins.index.AsyncIndexUpdate; +import org.apache.jackrabbit.oak.plugins.index.CompositeIndexEditorProvider; +import org.apache.jackrabbit.oak.plugins.index.IndexEditorProvider; +import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback; +import org.apache.jackrabbit.oak.plugins.index.counter.NodeCounterEditorProvider; +import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier; +import org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexDefinition; +import org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexEditorProvider; +import org.apache.jackrabbit.oak.plugins.index.lucene.util.IndexDefinitionBuilder; +import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeStore; +import org.apache.jackrabbit.oak.spi.commit.*; +import org.apache.jackrabbit.oak.spi.security.OpenSecurityProvider; +import org.apache.jackrabbit.oak.spi.state.NodeBuilder; +import org.apache.jackrabbit.oak.spi.state.NodeState; +import org.apache.jackrabbit.oak.spi.state.NodeStore; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FilterDirectory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.store.IndexOutput; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.apache.jackrabbit.oak.InitialContentHelper.INITIAL_CONTENT; +import static org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants.INDEX_DATA_CHILD_NAME; +import static org.junit.Assert.*; + +public class FailedIndexUpdateTest { + @Rule + public TemporaryFolder temporaryFolder = new TemporaryFolder(new File("target")); + + private Closer closer; + + private Root root; + private AsyncIndexUpdate asyncIndexUpdate; + private LocalDirectoryTrackingIndexCopier copier; + private FailIfDefinedEditorProvider luceneEditorProvider; + private FailOnDemandValidatorProvider failOnDemandValidatorProvider; + + private static final String TEST_CONTENT_PATH = "/test"; + + @Before + public void setup() throws Exception { + closer = Closer.create(); + createRepository(); + } + + private void createRepository() throws IOException { + ExecutorService executorService = Executors.newFixedThreadPool(2); + closer.register(new ExecutorCloser(executorService)); + copier = new LocalDirectoryTrackingIndexCopier(executorService, temporaryFolder.getRoot()); + luceneEditorProvider = new FailIfDefinedEditorProvider(copier); + + IndexEditorProvider editorProvider = new CompositeIndexEditorProvider(new NodeCounterEditorProvider(), luceneEditorProvider); + + NodeStore store = new MemoryNodeStore(INITIAL_CONTENT); + + Oak oak = new Oak(store) + .with(new OpenSecurityProvider()) + ; + root = oak.createRoot(); + + failOnDemandValidatorProvider = new FailOnDemandValidatorProvider(); + asyncIndexUpdate = new AsyncIndexUpdate("async", store, editorProvider); + asyncIndexUpdate.setValidatorProviders(Collections.singletonList(failOnDemandValidatorProvider)); + } + + @After + public void after() throws IOException { + closer.close(); + } + + @Test + public void workingReindexDirCleanUpOnFailureOfOtherIndex() throws Exception { + createIndex("fails", "foo", true); + asyncIndexUpdate.run(); + assertFalse("Indexing mustn't be failing", asyncIndexUpdate.isFailing()); + copier.clearStats(); + + createIndex("reindexing", "foo", false); + + root.getTree("/").addChild("test") + .addChild("a").setProperty("foo", "bar"); + root.commit(); + + asyncIndexUpdate.run(); + assertTrue("Indexing must fail", asyncIndexUpdate.isFailing()); + + Set reindexingDirPaths = copier.getReindexingDirPaths(); + assertEquals(1, reindexingDirPaths.size()); + + File reindexingDir = reindexingDirPaths.iterator().next(); + assertFalse("Reindexing directories must get cleaned up on failure", reindexingDir.exists()); + } + + @Test + public void workingReindexDirCleanUpOnFailureOfMerge() throws Exception { + failOnDemandValidatorProvider.shouldFail = true; + + createIndex("reindexing", "foo", false); + + root.getTree("/").addChild("test") + .addChild("a").setProperty("foo", "bar"); + root.commit(); + + asyncIndexUpdate.run(); + assertTrue("Indexing must fail", asyncIndexUpdate.isFailing()); + + Set reindexingDirPaths = copier.getReindexingDirPaths(); + assertEquals(1, reindexingDirPaths.size()); + + File reindexingDir = reindexingDirPaths.iterator().next(); + assertFalse("Reindexing directories must get cleaned up on failure", reindexingDir.exists()); + } + + private void createIndex(String idxName, String propName, boolean shouldFail) throws CommitFailedException { + IndexDefinitionBuilder idxBuilder = new IndexDefinitionBuilder(); + + idxBuilder + .includedPaths(TEST_CONTENT_PATH) + .indexRule("nt:base") + .property(propName).propertyIndex(); + Tree idx = idxBuilder.build(root.getTree("/oak:index").addChild(idxName)); + idx.setProperty("shouldFail", shouldFail); + + root.commit(); + } + + static class LocalDirectoryTrackingIndexCopier extends IndexCopier { + + private final Map dirs = Maps.newHashMap(); + private final Map dirPaths = Maps.newHashMap(); + private final Set reindexingDirPaths = Sets.newHashSet(); + + LocalDirectoryTrackingIndexCopier(Executor executor, File indexRootDir) throws IOException { + super(executor, indexRootDir); + } + + @Override + public CreateFileTrackingDirectory wrapForWrite(LuceneIndexDefinition definition, + Directory remote, + boolean reindexMode, String dirName, + COWDirecetoryTracker cowDirecetoryTracker) throws IOException { + CreateFileTrackingDirectory dir = new CreateFileTrackingDirectory( + super.wrapForWrite(definition, remote, reindexMode, dirName, cowDirecetoryTracker)); + + String indexPath = definition.getIndexPath(); + dirs.put(indexPath, dir); + File dirPath = getIndexDir(definition, indexPath, dirName); + dirPaths.put(indexPath, dirPath); + if (reindexMode) { + reindexingDirPaths.add(dirPath); + } + + return dir; + } + + void clearStats() { + dirs.clear(); + dirPaths.clear(); + reindexingDirPaths.clear(); + } + + Map getDirs() { + return dirs; + } + + Map getDirPaths() { + return dirPaths; + } + + Set getReindexingDirPaths() { + return reindexingDirPaths; + } + } + + static class CreateFileTrackingDirectory extends FilterDirectory { + + private final Set createdFiles; + private boolean closed; + + CreateFileTrackingDirectory(Directory in) { + super(in); + createdFiles = Sets.newHashSet(); + closed = false; + } + + @Override + public IndexOutput createOutput(String name, IOContext context) throws IOException { + createdFiles.add(name); + return super.createOutput(name, context); + } + + @Override + public void close() throws IOException { + super.close(); + closed = true; + } + + Set getCreatedFiles() { + return createdFiles; + } + + boolean isClosed() { + return closed; + } + } + + private static class FailIfDefinedEditorProvider extends LuceneIndexEditorProvider { + FailIfDefinedEditorProvider(IndexCopier copier) { + super(copier); + } + + @Override + public Editor getIndexEditor(@NotNull String type, @NotNull NodeBuilder definition, + @NotNull NodeState root, + @NotNull IndexUpdateCallback callback) throws CommitFailedException { + Editor editor = super.getIndexEditor(type, definition, root, callback); + if (definition.getBoolean("shouldFail")) { + editor = new FailOnLeavePathEditor(editor, TEST_CONTENT_PATH); + } + return editor; + } + } + + private static class FailOnLeavePathEditor implements Editor { + private final Editor delegate; + private final String failingPath; + final String currPath; + + FailOnLeavePathEditor(Editor delegate, String failingPath) { + this(delegate, failingPath, "", ""); + } + + private FailOnLeavePathEditor(Editor delegate, String failingPath, String parentPath, String name) { + this.delegate = delegate != null ? delegate : new DefaultEditor(); + this.failingPath = failingPath; + this.currPath = ("/".equals(parentPath) ? parentPath : parentPath + "/") + name; + } + + @Override + public void enter(NodeState before, NodeState after) throws CommitFailedException { + delegate.enter(before, after); + } + + @Override + public void leave(NodeState before, NodeState after) throws CommitFailedException { + delegate.leave(before, after); // delegate call before failing + + if (failingPath.equals(currPath)) { + throw new CommitFailedException("index-fail", 1, null); + } + } + + @Override + public void propertyAdded(PropertyState after) throws CommitFailedException { + delegate.propertyAdded(after); + } + + @Override + public void propertyChanged(PropertyState before, PropertyState after) throws CommitFailedException { + delegate.propertyChanged(before, after); + } + + @Override + public void propertyDeleted(PropertyState before) throws CommitFailedException { + delegate.propertyDeleted(before); + } + + @Override + @Nullable + public Editor childNodeAdded(String name, NodeState after) throws CommitFailedException { + return new FailOnLeavePathEditor(delegate.childNodeAdded(name, after), failingPath, currPath, name); + } + + @Override + @Nullable + public Editor childNodeChanged(String name, NodeState before, NodeState after) throws CommitFailedException { + return new FailOnLeavePathEditor(delegate.childNodeChanged(name, before, after), failingPath, currPath, name); + } + + @Override + @Nullable + public Editor childNodeDeleted(String name, NodeState before) throws CommitFailedException { + return new FailOnLeavePathEditor(delegate.childNodeDeleted(name, before), failingPath, currPath, name); + } + } + + static class FailOnDemandValidatorProvider extends ValidatorProvider { + + boolean shouldFail; + static final String FAILING_PATH_FRAGMENT = INDEX_DATA_CHILD_NAME; + + @Override + protected @Nullable Validator getRootValidator(NodeState before, NodeState after, CommitInfo info) { + return new FailOnDemandValidator(new DefaultValidator()); + } + + class FailOnDemandValidator extends FailOnLeavePathEditor implements Validator { + final Validator delegate; + + FailOnDemandValidator(Validator delegate) { + super(delegate, ""); + this.delegate = delegate; + } + + private FailOnDemandValidator(Validator delegate, String parentPath, String name) { + super(delegate, "", parentPath, name); + this.delegate = delegate != null ? delegate : new DefaultValidator(); + } + + @Override + public void leave(NodeState before, NodeState after) throws CommitFailedException { + super.leave(before, after); + + if (shouldFail && currPath.contains(FAILING_PATH_FRAGMENT)) { + throw new CommitFailedException("validator-fail", 1, null); + } + } + + @Override + @Nullable + public Validator childNodeAdded(String name, NodeState after) throws CommitFailedException { + return new FailOnDemandValidator(delegate.childNodeAdded(name, after), currPath, name); + } + + @Override + @Nullable + public Validator childNodeChanged(String name, NodeState before, NodeState after) throws CommitFailedException { + return new FailOnDemandValidator(delegate.childNodeChanged(name, before, after), currPath, name); + } + + @Override + @Nullable + public Validator childNodeDeleted(String name, NodeState before) throws CommitFailedException { + return new FailOnDemandValidator(delegate.childNodeDeleted(name, before), currPath, name); + } + } + } +} \ No newline at end of file