You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by mk...@apache.org on 2022/03/28 10:08:27 UTC
[jackrabbit-oak] branch trunk updated: OAK-9721: Add state information while importing index (#527)
This is an automated email from the ASF dual-hosted git repository.
mkataria pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
The following commit(s) were added to refs/heads/trunk by this push:
new 42f1754 OAK-9721: Add state information while importing index (#527)
42f1754 is described below
commit 42f1754e5756209da47299b9785f176948c885b7
Author: Mohit Kataria <mk...@apache.org>
AuthorDate: Mon Mar 28 15:38:20 2022 +0530
OAK-9721: Add state information while importing index (#527)
* OAK-9721: Add state information while importing index
---
.../oak/plugins/index/importer/IndexImporter.java | 261 ++++++++++++++++-----
.../oak/plugins/index/importer/package-info.java | 2 +-
.../plugins/index/importer/IndexImporterTest.java | 180 +++++++++++++-
.../lucene/LuceneIndexImporterReindexTest.java | 89 +++++++
.../index/elastic/ElasticIndexImporter.java | 4 +
.../plugins/index/search/ReindexOperations.java | 3 +
.../plugins/index/IndexImporterReindexTest.java | 62 +++++
7 files changed, 534 insertions(+), 67 deletions(-)
diff --git a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporter.java b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporter.java
index fb86df7..06d8ba8 100644
--- a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporter.java
+++ b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporter.java
@@ -22,8 +22,10 @@ package org.apache.jackrabbit.oak.plugins.index.importer;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
@@ -63,7 +65,7 @@ public class IndexImporter {
*/
public static final String OAK_INDEX_IMPORTER_PRESERVE_CHECKPOINT = "oak.index.importer.preserveCheckpoint";
- private final Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(IndexImporter.class);
private final NodeStore nodeStore;
private final File indexDir;
private final Map<String, IndexImporterProvider> importers = new HashMap<>();
@@ -76,6 +78,10 @@ public class IndexImporter {
private final IndexDefinitionUpdater indexDefinitionUpdater;
private final boolean preserveCheckpoint = Boolean.getBoolean(OAK_INDEX_IMPORTER_PRESERVE_CHECKPOINT);
+ static final int RETRIES = Integer.getInteger("oak.index.import.retries", 5);
+ public static final String INDEX_IMPORT_STATE_KEY = "indexImportState";
+ private final Set<String> indexPathsToUpdate;
+
public IndexImporter(NodeStore nodeStore, File indexDir, IndexEditorProvider indexEditorProvider,
AsyncIndexerLock indexerLock) throws IOException {
checkArgument(indexDir.exists() && indexDir.isDirectory(), "Path [%s] does not point " +
@@ -90,72 +96,136 @@ public class IndexImporter {
"checkpointed state [%s]", indexerInfo.checkpoint);
this.indexDefinitionUpdater = new IndexDefinitionUpdater(new File(indexDir, INDEX_DEFINITIONS_JSON));
this.asyncLaneToIndexMapping = mapIndexesToLanes(indexes);
+ this.indexPathsToUpdate = new HashSet<>();
+ }
+
+ enum IndexImportState {
+ NULL, SWITCH_LANE, IMPORT_INDEX_DATA, BRING_INDEX_UPTODATE, RELEASE_CHECKPOINT
}
public void importIndex() throws IOException, CommitFailedException {
- if (indexes.keySet().isEmpty()) {
- log.warn("No indexes to import (possibly index definitions outside of a oak:index node?)");
+ try {
+ if (indexes.keySet().isEmpty()) {
+ LOG.warn("No indexes to import (possibly index definitions outside of a oak:index node?)");
+ }
+ LOG.info("Proceeding to import {} indexes from {}", indexes.keySet(), indexDir.getAbsolutePath());
+
+ //TODO Need to review it for idempotent design. A failure in any step should not
+ //leave setup in in consistent state and provide option for recovery
+
+ //Step 1 - Switch the index lanes so that async indexer does not touch them
+ //while we are importing the index data
+ runWithRetry(RETRIES, IndexImportState.SWITCH_LANE, this::switchLanes);
+ LOG.info("Done with switching of index lanes before import");
+
+ //Step 2 - Import the existing index data.
+ // In this step we are:
+ // switching lane for new index
+ // incrementing reindex count.
+ // marking index as disabled in case of superseded index
+ // after this step new index is available in repository
+ runWithRetry(RETRIES, IndexImportState.IMPORT_INDEX_DATA, this::importIndexData);
+ LOG.info("Done with importing of index data");
+
+ //Step 3 - Bring index upto date.
+ // In this step we are:
+ // interrupting current indexing.
+ // reverting lane back to async
+ // resuming current indexing;
+ runWithRetry(RETRIES, IndexImportState.BRING_INDEX_UPTODATE, this::bringIndexUpToDate);
+ LOG.info("Done with bringing index up-to-date");
+ //Step 4 - Release the checkpoint
+ // this is again an idempotent function
+ runWithRetry(RETRIES, IndexImportState.RELEASE_CHECKPOINT, this::releaseCheckpoint);
+ LOG.info("Done with releasing checkpoint");
+
+ // Remove indexImportState property on successful import. In case of preserveCheckpoint is enabled,
+ // we assume that index import is done without releaseCheckpoint. In that case this method below will be NOOP
+ // as currentIndexImportState is null and doesn't match RELEASE_CHECKPOINT state
+ updateIndexImporterState(IndexImportState.RELEASE_CHECKPOINT, null, true);
+ LOG.info("Done with removing index import state");
+
+ } catch (CommitFailedException | IOException e) {
+
+ LOG.error("Failure while index import", e);
+ try {
+ runWithRetry(RETRIES, null, () -> {
+ NodeState root = nodeStore.getRoot();
+ NodeBuilder builder = root.builder();
+ revertLaneChange(builder, indexPathsToUpdate);
+ mergeWithConcurrentCheck(nodeStore, builder);
+ });
+ } catch (CommitFailedException commitFailedException) {
+ LOG.error("Unable to revert back index lanes for: "
+ + indexPathsToUpdate.stream().collect(StringBuilder::new, StringBuilder::append,
+ (a, b) -> a.append(",").append(b)).toString(), commitFailedException);
+ throw e;
+ }
}
- log.info("Proceeding to import {} indexes from {}", indexes.keySet(), indexDir.getAbsolutePath());
-
- //TODO Need to review it for idempotent design. A failure in any step should not
- //leave setup in in consistent state and provide option for recovery
-
- //Step 1 - Switch the index lanes so that async indexer does not touch them
- //while we are importing the index data
- switchLanes();
- log.info("Done with switching of index lanes before import");
-
- //Step 2 - Import the existing index data
- importIndexData();
- log.info("Done with importing of index data");
-
- //Step 3 - Bring index upto date
- bringIndexUpToDate();
-
- //Step 4 - Release the checkpoint
- releaseCheckpoint();
}
public void addImporterProvider(IndexImporterProvider importerProvider) {
importers.put(importerProvider.getType(), importerProvider);
}
- void switchLanes() throws CommitFailedException, IOException {
- NodeState root = nodeStore.getRoot();
- NodeBuilder builder = root.builder();
-
- for (IndexInfo indexInfo : asyncLaneToIndexMapping.values()){
- if (!indexInfo.newIndex) {
- NodeBuilder idxBuilder = NodeStoreUtils.childBuilder(builder, indexInfo.indexPath);
- AsyncLaneSwitcher.switchLane(idxBuilder, AsyncLaneSwitcher.getTempLaneName(indexInfo.asyncLaneName));
+ void switchLanes() throws CommitFailedException {
+ try {
+ NodeState root = nodeStore.getRoot();
+ NodeBuilder builder = root.builder();
+
+ for (IndexInfo indexInfo : asyncLaneToIndexMapping.values()) {
+ if (!indexInfo.newIndex) {
+ NodeBuilder idxBuilder = NodeStoreUtils.childBuilder(builder, indexInfo.indexPath);
+ indexPathsToUpdate.add(indexInfo.indexPath);
+ AsyncLaneSwitcher.switchLane(idxBuilder, AsyncLaneSwitcher.getTempLaneName(indexInfo.asyncLaneName));
+ }
}
+ updateIndexImporterState(builder, IndexImportState.NULL, IndexImportState.SWITCH_LANE, false);
+ mergeWithConcurrentCheck(nodeStore, builder);
+ } catch (CommitFailedException e) {
+ LOG.error("Failed while performing switchLanes and updating indexImportState from [{}] to [{}]",
+ IndexImportState.NULL, IndexImportState.SWITCH_LANE);
+ throw e;
}
- mergeWithConcurrentCheck(nodeStore, builder);
}
void importIndexData() throws CommitFailedException, IOException {
- NodeState root = nodeStore.getRoot();
- NodeBuilder rootBuilder = root.builder();
- IndexDisabler indexDisabler = new IndexDisabler(rootBuilder);
- for (IndexInfo indexInfo : asyncLaneToIndexMapping.values()) {
- log.info("Importing index data for {}", indexInfo.indexPath);
- NodeBuilder idxBuilder = indexDefinitionUpdater.apply(rootBuilder, indexInfo.indexPath);
-
- if (indexInfo.newIndex) {
- AsyncLaneSwitcher.switchLane(idxBuilder, AsyncLaneSwitcher.getTempLaneName(indexInfo.asyncLaneName));
- } else {
- //For existing ind
- NodeState existing = NodeStateUtils.getNode(root, indexInfo.indexPath);
- copyLaneProps(existing, idxBuilder);
+ try {
+ NodeState root = nodeStore.getRoot();
+ NodeBuilder rootBuilder = root.builder();
+ IndexDisabler indexDisabler = new IndexDisabler(rootBuilder);
+ for (IndexInfo indexInfo : asyncLaneToIndexMapping.values()) {
+ LOG.info("Importing index data for {}", indexInfo.indexPath);
+ // current index node contains : temp-async and async-previous.
+ // old state is indexdefinition with async=async
+ // indexDefinitionUpdater contains base index-def i.e. without async-previous and temp-async
+ // this apply method take current state and return Nodebuilder with old indexdefinitions i.e. without async-temp.
+ // rootbuilder also get updated in the process
+ NodeBuilder idxBuilder = indexDefinitionUpdater.apply(rootBuilder, indexInfo.indexPath);
+
+ if (indexInfo.newIndex) {
+ AsyncLaneSwitcher.switchLane(idxBuilder, AsyncLaneSwitcher.getTempLaneName(indexInfo.asyncLaneName));
+ indexPathsToUpdate.add(indexInfo.indexPath);
+ } else {
+ //For existing index
+ NodeState existing = NodeStateUtils.getNode(root, indexInfo.indexPath);
+ // copyLaneProps copies property values for async and previous-async property that we set in method
+ // "switchLanes" to idxBuilder from indexDefinitionUpdater i.e. nodestate before index import started
+ copyLaneProps(existing, idxBuilder);
+ }
+ //TODO How to support CompositeNodeStore where some of the child nodes would be hidden
+ incrementReIndexCount(idxBuilder);
+ // importIndex copies data from current folder to new older. Updates idxbuilder with new uid.
+ getImporter(indexInfo.type).importIndex(root, idxBuilder, indexInfo.indexDir);
+ indexDisabler.markDisableFlagIfRequired(indexInfo.indexPath, idxBuilder);
}
- //TODO How to support CompositeNodeStore where some of the child nodes would be hidden
- incrementReIndexCount(idxBuilder);
- getImporter(indexInfo.type).importIndex(root, idxBuilder, indexInfo.indexDir);
-
- indexDisabler.markDisableFlagIfRequired(indexInfo.indexPath, idxBuilder);
+ updateIndexImporterState(root.builder(), IndexImportState.SWITCH_LANE, IndexImportState.IMPORT_INDEX_DATA, false);
+ mergeWithConcurrentCheck(nodeStore, rootBuilder, indexEditorProvider);
+ } catch (CommitFailedException e) {
+ LOG.error("Failed while performing importIndexData and updating indexImportState from [{}] to [{}]",
+ IndexImportState.SWITCH_LANE, IndexImportState.IMPORT_INDEX_DATA);
+ throw e;
}
- mergeWithConcurrentCheck(nodeStore, rootBuilder, indexEditorProvider);
}
private void bringIndexUpToDate() throws CommitFailedException {
@@ -178,7 +248,7 @@ public class IndexImporter {
NodeState after = nodeStore.retrieve(checkpoint);
checkNotNull(after, "No state found for checkpoint [%s] for lane [%s]",checkpoint, laneName);
- log.info("Proceeding to update imported indexes {} to checkpoint [{}] for lane [{}]",
+ LOG.info("Proceeding to update imported indexes {} to checkpoint [{}] for lane [{}]",
indexInfos, checkpoint, laneName);
NodeState before = indexedState;
@@ -201,23 +271,27 @@ public class IndexImporter {
}
revertLaneChange(builder, indexInfos);
-
+ updateIndexImporterState(builder, IndexImportState.IMPORT_INDEX_DATA, IndexImportState.BRING_INDEX_UPTODATE, false);
mergeWithConcurrentCheck(nodeStore, builder);
success = true;
- log.info("Imported index is updated to repository state at checkpoint [{}] for " +
+ LOG.info("Imported index is updated to repository state at checkpoint [{}] for " +
"indexing lane [{}]", checkpoint, laneName);
+ } catch (CommitFailedException e) {
+ LOG.error("Failed while performing bringIndexUpToDate and updating indexImportState from [{}] to [{}]",
+ IndexImportState.IMPORT_INDEX_DATA, IndexImportState.BRING_INDEX_UPTODATE);
+ throw e;
} finally {
try {
resumeCurrentIndexing(lockToken);
} catch (CommitFailedException | RuntimeException e) {
- log.warn("Error occurred while releasing indexer lock", e);
+ LOG.warn("Error occurred while releasing indexer lock", e);
if (success) {
throw e;
}
}
}
- log.info("Import done for indexes {}", indexInfos);
+ LOG.info("Import done for indexes {}", indexInfos);
}
private void revertLaneChange(NodeBuilder builder, List<IndexInfo> indexInfos) {
@@ -227,6 +301,47 @@ public class IndexImporter {
}
}
+ private void revertLaneChange(NodeBuilder builder, Set<String> indexPaths) {
+ for (String indexPath : indexPaths) {
+ NodeBuilder idxBuilder = NodeStoreUtils.childBuilder(builder, indexPath);
+ AsyncLaneSwitcher.revertSwitch(idxBuilder, indexPath);
+ }
+ }
+
+ private IndexImportState getIndexImportState(NodeBuilder nodeBuilder) {
+ if (nodeBuilder.getProperty(INDEX_IMPORT_STATE_KEY) == null || nodeBuilder.getProperty(INDEX_IMPORT_STATE_KEY).getValue(Type.STRING) == null) {
+ return IndexImportState.NULL;
+ } else {
+ return IndexImportState.valueOf(nodeBuilder.getProperty(INDEX_IMPORT_STATE_KEY).getValue(Type.STRING));
+ }
+ }
+
+ // updateIndexImporterState is an idempotent process and only updates state if currentImportState matches.
+ private void updateIndexImporterState(IndexImportState currentImportState, IndexImportState nextImportState, boolean shouldCommit) throws CommitFailedException {
+ NodeState root = nodeStore.getRoot();
+ NodeBuilder builder = root.builder();
+ updateIndexImporterState(builder, currentImportState, nextImportState, shouldCommit);
+ }
+
+ // updateIndexImporterState is an idempotent process and only updates state if currentImportState matches nodeStoreIndexImportState.
+ private void updateIndexImporterState(NodeBuilder builder, IndexImportState currentImportState, IndexImportState nextImportState, boolean shouldCommit) throws CommitFailedException {
+ for (String indexPath : indexPathsToUpdate) {
+ NodeBuilder idxBuilder = NodeStoreUtils.childBuilder(builder, indexPath);
+ IndexImportState nodeStoreIndexImportState = getIndexImportState(idxBuilder);
+
+ if (nodeStoreIndexImportState == currentImportState) {
+ if (nextImportState == IndexImportState.NULL) {
+ idxBuilder.removeProperty(INDEX_IMPORT_STATE_KEY);
+ } else {
+ idxBuilder.setProperty(INDEX_IMPORT_STATE_KEY, nextImportState.toString(), Type.STRING);
+ }
+ }
+ }
+ if (shouldCommit) {
+ mergeWithConcurrentCheck(nodeStore, builder);
+ }
+ }
+
private void resumeCurrentIndexing(LockToken lockToken) throws CommitFailedException {
indexerLock.unlock(lockToken);
}
@@ -282,7 +397,7 @@ public class IndexImporter {
* Determines the async lane name. This method also check if lane was previously switched
* then it uses the actual lane name prior to switch was done
*
- * @param indexPath path of index. Mostly used in reporting exception
+ * @param indexPath path of index. Mostly used in reporting exception
* @param indexState nodeState for index at given path
*
* @return async lane name or null which would be the case for sync indexes
@@ -295,15 +410,18 @@ public class IndexImporter {
return IndexUtils.getAsyncLaneName(indexState, indexPath);
}
- private void releaseCheckpoint() {
+ private void releaseCheckpoint() throws CommitFailedException {
if (preserveCheckpoint) {
- log.info("Preserving the referred checkpoint [{}]. This could have been done in case this checkpoint is needed by a process later on." +
+ LOG.info("Preserving the referred checkpoint [{}]. This could have been done in case this checkpoint is needed by a process later on." +
" Please make sure to remove the checkpoint once it's no longer needed.", indexerInfo.checkpoint);
+ // We are assuming that indexImport is complete from our end as checkpoint need to be preserved
+ updateIndexImporterState(IndexImportState.BRING_INDEX_UPTODATE, null, true);
} else {
- nodeStore.release(indexerInfo.checkpoint);
- log.info("Released the referred checkpoint [{}]", indexerInfo.checkpoint);
+ if (nodeStore.release(indexerInfo.checkpoint)) {
+ LOG.info("Released the referred checkpoint [{}]", indexerInfo.checkpoint);
+ updateIndexImporterState(IndexImportState.BRING_INDEX_UPTODATE, IndexImportState.RELEASE_CHECKPOINT, true);
+ }
}
-
}
private void incrementReIndexCount(NodeBuilder definition) {
@@ -339,4 +457,25 @@ public class IndexImporter {
}
}
+ interface IndexImporterStepExecutor {
+ void execute() throws CommitFailedException, IOException;
+ }
+
+ void runWithRetry(int maxRetries, IndexImportState indexImportState, IndexImporterStepExecutor step) throws CommitFailedException, IOException {
+ int count = 1;
+ while (count <= maxRetries) {
+ LOG.info("IndexImporterStepExecutor:{} ,count:{}", indexImportState, count);
+ try {
+ step.execute();
+ break;
+ } catch (CommitFailedException | IOException e) {
+ LOG.warn("IndexImporterStepExecutor:{} fail count: {}, retries left: {}", indexImportState, count, maxRetries - count, e);
+ if (count++ >= maxRetries) {
+ LOG.warn("IndexImporterStepExecutor:{} failed after {} retries", indexImportState, maxRetries, e);
+ throw e;
+ }
+ }
+ }
+ }
+
}
diff --git a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/package-info.java b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/package-info.java
index 2ebba64..6590336 100644
--- a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/package-info.java
+++ b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/importer/package-info.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-@Version("0.2.0")
+@Version("0.3.0")
package org.apache.jackrabbit.oak.plugins.index.importer;
import org.osgi.annotation.versioning.Version;
\ No newline at end of file
diff --git a/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporterTest.java b/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporterTest.java
index db28d42..88d9614 100644
--- a/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporterTest.java
+++ b/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/importer/IndexImporterTest.java
@@ -23,6 +23,7 @@ import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
+import java.text.MessageFormat;
import java.util.Properties;
import java.util.Set;
@@ -33,6 +34,7 @@ import org.apache.felix.inventory.Format;
import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.api.Type;
import org.apache.jackrabbit.oak.commons.PathUtils;
+import org.apache.jackrabbit.oak.commons.junit.LogCustomizer;
import org.apache.jackrabbit.oak.plugins.index.AsyncIndexUpdate;
import org.apache.jackrabbit.oak.plugins.index.IndexConstants;
import org.apache.jackrabbit.oak.plugins.index.IndexEditorProvider;
@@ -64,6 +66,7 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.slf4j.event.Level;
import static com.google.common.base.Charsets.UTF_8;
import static com.google.common.collect.ImmutableSet.of;
@@ -89,6 +92,19 @@ public class IndexImporterTest {
private NodeStore store = new MemoryNodeStore();
private IndexEditorProvider provider = new PropertyIndexEditorProvider();
+ private LogCustomizer customizer;
+
+ @Before
+ public void setup(){
+ customizer = LogCustomizer.forLogger(IndexImporter.class.getName()).filter(Level.INFO).create();
+ customizer.starting();
+ }
+
+ @After
+ public void after() {
+ customizer.finished();
+ }
+
@Test(expected = IllegalArgumentException.class)
public void importIndex_NoMeta() throws Exception{
@@ -143,10 +159,11 @@ public class IndexImporterTest {
NodeBuilder builder = store.getRoot().builder();
builder.child("idx-a").setProperty("type", "property");
builder.child("idx-a").setProperty("foo", "bar");
+ builder.child("idx-a").setProperty("async", "async");
store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
- createIndexDirs("/idx-a");
+ String checkpoint = createIndexDirs("/idx-a");
IndexImporter importer = new IndexImporter(store, temporaryFolder.getRoot(), provider, NOOP_LOCK);
@@ -164,10 +181,11 @@ public class IndexImporterTest {
}
};
importer.addImporterProvider(provider);
- importer.switchLanes();
- importer.importIndexData();
-
- assertTrue(store.getRoot().getChildNode("idx-a").getBoolean("imported"));
+ try{
+ importer.importIndex();
+ } catch (Exception e){
+ assertTrue(store.getRoot().getChildNode("idx-a").getBoolean("imported"));
+ }
}
@Test
@@ -264,6 +282,7 @@ public class IndexImporterTest {
//It would not pickup /e as thats not yet indexed as part of last checkpoint
assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
assertNull(store.retrieve(checkpoint));
+ assertNull(store.getRoot().getChildNode("oak:index").getChildNode("fooIndex").getProperty(IndexImporter.INDEX_IMPORT_STATE_KEY));
}
@Test
@@ -538,4 +557,155 @@ public class IndexImporterTest {
printer.print(pw, Format.JSON, false);
Files.write(sw.toString(), file, UTF_8);
}
+
+ private String importDataIncrementalUpdateBeforeSetupMethod() throws IOException, CommitFailedException {
+ NodeBuilder builder = store.getRoot().builder();
+ createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME),
+ "fooIndex", true, false, ImmutableSet.of("foo"), null)
+ .setProperty(ASYNC_PROPERTY_NAME, "async");
+ builder.child("a").setProperty("foo", "abc");
+ builder.child("b").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ new AsyncIndexUpdate("async", store, provider).run();
+
+ String checkpoint = createIndexDirs("/oak:index/fooIndex");
+ builder = store.getRoot().builder();
+ builder.child("c").setProperty("foo", "abc");
+ builder.child("d").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ new AsyncIndexUpdate("async", store, provider).run();
+
+ FilterImpl f = createFilter(store.getRoot(), NT_BASE);
+ PropertyIndexLookup lookup = new PropertyIndexLookup(store.getRoot());
+ assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
+
+ builder = store.getRoot().builder();
+ builder.child("e").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ return checkpoint;
+ }
+
+ private IndexImporterProvider getImporterProvider(String checkpoint) {
+ IndexImporterProvider importerProvider = new IndexImporterProvider() {
+ @Override
+ public void importIndex(NodeState root, NodeBuilder defn, File indexDir) {
+ assertEquals("fooIndex", indexDir.getName());
+ assertEquals(2, defn.getProperty(REINDEX_COUNT).getValue(Type.LONG).longValue());
+ defn.getChildNode(IndexConstants.INDEX_CONTENT_NODE_NAME).remove();
+
+ NodeState cpState = store.retrieve(checkpoint);
+ NodeState indexData = NodeStateUtils.getNode(cpState, "/oak:index/fooIndex/:index");
+ defn.setChildNode(IndexConstants.INDEX_CONTENT_NODE_NAME, indexData);
+ }
+
+ @Override
+ public String getType() {
+ return "property";
+ }
+ };
+ return importerProvider;
+ }
+
+ @Test
+ public void importDataIncrementalupdateTestFailureAtSwitchlaneState() throws Exception {
+ String checkpoint = importDataIncrementalUpdateBeforeSetupMethod();
+ try {
+ IndexImporter importer = new IndexImporter(store, temporaryFolder.getRoot(), provider, NOOP_LOCK) {
+ @Override
+ void switchLanes() throws CommitFailedException {
+ throw new CommitFailedException("dummy", 1, "Explicitly throw CommitFailedException");
+ }
+ };
+ importer.addImporterProvider(getImporterProvider(checkpoint));
+ importer.importIndex();
+ } finally {
+ NodeState idx = store.getRoot().getChildNode("oak:index").getChildNode("fooIndex");
+ assertEquals("async", idx.getString("async"));
+ FilterImpl f = createFilter(store.getRoot(), NT_BASE);
+ PropertyIndexLookup lookup = new PropertyIndexLookup(store.getRoot());
+ assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
+
+ lookup = new PropertyIndexLookup(store.getRoot());
+ //It would not pickup /e as thats not yet indexed as part of last checkpoint
+ assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
+ // checkpoint is not released because of failure
+ assertNotNull(store.retrieve(checkpoint));
+ // As failure is on switchlanes no update happened i.e. no state change got committed
+ assertEquals("State matching failed",
+ null,
+ store.getRoot().getChildNode("oak:index").getChildNode("fooIndex").getProperty(IndexImporter.INDEX_IMPORT_STATE_KEY));
+
+ // Test retry logic
+ String failureLog = MessageFormat.format("IndexImporterStepExecutor:{0} failed after {1} retries",
+ IndexImporter.IndexImportState.SWITCH_LANE, IndexImporter.RETRIES);
+ boolean failureLogPresent = false;
+ for (String log : customizer.getLogs()) {
+ if (log.equals(failureLog)) {
+ failureLogPresent = true;
+ break;
+ }
+ }
+ assertTrue(failureLogPresent);
+
+ assertEquals("State matching failed",
+ null,
+ store.getRoot().getChildNode("oak:index").getChildNode("fooIndex").getProperty(IndexImporter.INDEX_IMPORT_STATE_KEY));
+ }
+ }
+
+ @Test
+ public void importDataIncrementalUpdateTestFailureAtImportIndexDataState() throws Exception {
+ String checkpoint = importDataIncrementalUpdateBeforeSetupMethod();
+ try {
+ IndexImporter importer = new IndexImporter(store, temporaryFolder.getRoot(), provider, NOOP_LOCK) {
+ @Override
+ void importIndexData() throws CommitFailedException, IOException {
+ throw new IOException("Explicitly throw IOException");
+ }
+ };
+ ;
+ importer.addImporterProvider(getImporterProvider(checkpoint));
+ importer.importIndex();
+ } finally {
+ NodeState idx = store.getRoot().getChildNode("oak:index").getChildNode("fooIndex");
+ assertEquals("async", idx.getString("async"));
+ FilterImpl f = createFilter(store.getRoot(), NT_BASE);
+ PropertyIndexLookup lookup = new PropertyIndexLookup(store.getRoot());
+ assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
+
+ lookup = new PropertyIndexLookup(store.getRoot());
+ //It would not pickup /e as thats not yet indexed as part of last checkpoint
+ assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
+ assertNotNull(store.retrieve(checkpoint));
+ assertEquals("State matching failed",
+ IndexImporter.IndexImportState.SWITCH_LANE.toString(),
+ store.getRoot().getChildNode("oak:index").getChildNode("fooIndex").getProperty(IndexImporter.INDEX_IMPORT_STATE_KEY).getValue(Type.STRING));
+ }
+ }
+
+ @Test
+ public void importDataIncrementalUpdateNoFailure() throws Exception {
+ String checkpoint = importDataIncrementalUpdateBeforeSetupMethod();
+ try {
+ IndexImporter importer = new IndexImporter(store, temporaryFolder.getRoot(), provider, NOOP_LOCK);
+ importer.addImporterProvider(getImporterProvider(checkpoint));
+ importer.importIndex();
+ NodeState idx = store.getRoot().getChildNode("oak:index").getChildNode("fooIndex");
+ assertEquals("async", idx.getString("async"));
+ } finally {
+ NodeState idx = store.getRoot().getChildNode("oak:index").getChildNode("fooIndex");
+ assertEquals("async", idx.getString("async"));
+ FilterImpl f = createFilter(store.getRoot(), NT_BASE);
+ PropertyIndexLookup lookup = new PropertyIndexLookup(store.getRoot());
+ //It would not pickup /e as thats not yet indexed as part of last checkpoint
+ assertEquals(of("a", "b", "c", "d"), find(lookup, "foo", "abc", f));
+ assertNull(store.retrieve(checkpoint));
+ assertEquals("State matching failed",
+ null,
+ store.getRoot().getChildNode("oak:index").getChildNode("fooIndex").getProperty(IndexImporter.INDEX_IMPORT_STATE_KEY));
+ }
+ }
}
diff --git a/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexImporterReindexTest.java b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexImporterReindexTest.java
new file mode 100644
index 0000000..51eefb3
--- /dev/null
+++ b/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexImporterReindexTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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 org.apache.jackrabbit.JcrConstants;
+import org.apache.jackrabbit.oak.InitialContentHelper;
+import org.apache.jackrabbit.oak.api.ContentRepository;
+import org.apache.jackrabbit.oak.api.Tree;
+import org.apache.jackrabbit.oak.api.Type;
+import org.apache.jackrabbit.oak.plugins.index.FunctionIndexCommonTest;
+import org.apache.jackrabbit.oak.plugins.index.IndexImporterReindexTest;
+import org.apache.jackrabbit.oak.plugins.index.LuceneIndexOptions;
+import org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants;
+import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeStore;
+import org.apache.jackrabbit.oak.plugins.memory.PropertyStates;
+import org.junit.After;
+import org.junit.Rule;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.INDEX_DEFINITIONS_NAME;
+import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.INDEX_DEFINITIONS_NODE_TYPE;
+import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.REINDEX_PROPERTY_NAME;
+import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_PROPERTY_NAME;
+
+public class LuceneIndexImporterReindexTest extends IndexImporterReindexTest {
+
+ private ExecutorService executorService = Executors.newFixedThreadPool(2);
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder(new File("target"));
+
+ protected Tree createIndex(String name, Set<String> propNames) {
+ Tree index = root.getTree("/");
+ return createIndex(index, name, propNames);
+ }
+
+ protected Tree createIndex(Tree index, String name, Set<String> propNames) {
+ Tree def = index.addChild(INDEX_DEFINITIONS_NAME).addChild(name);
+ def.setProperty(JcrConstants.JCR_PRIMARYTYPE,
+ INDEX_DEFINITIONS_NODE_TYPE, Type.NAME);
+ def.setProperty(TYPE_PROPERTY_NAME, LuceneIndexConstants.TYPE_LUCENE);
+ def.setProperty(REINDEX_PROPERTY_NAME, true);
+ def.setProperty(FulltextIndexConstants.FULL_TEXT_ENABLED, false);
+ def.setProperty(PropertyStates.createProperty(FulltextIndexConstants.INCLUDE_PROPERTY_NAMES, propNames, Type.STRINGS));
+ def.setProperty(LuceneIndexConstants.SAVE_DIR_LISTING, true);
+ return index.getChild(INDEX_DEFINITIONS_NAME).getChild(name);
+ }
+
+ @Override
+ protected ContentRepository createRepository() {
+ LuceneTestRepositoryBuilder luceneTestRepositoryBuilder = new LuceneTestRepositoryBuilder(executorService, temporaryFolder);
+ luceneTestRepositoryBuilder.setNodeStore(new MemoryNodeStore(InitialContentHelper.INITIAL_CONTENT));
+ repositoryOptionsUtil = luceneTestRepositoryBuilder.build();
+ indexOptions = new LuceneIndexOptions();
+ return repositoryOptionsUtil.getOak()
+ .createContentRepository();
+ }
+
+ @Override
+ protected String getLoggerName() {
+ return LuceneIndexEditor.class.getName();
+ }
+
+ @After
+ public void shutdownExecutor() {
+ executorService.shutdown();
+ }
+
+}
diff --git a/oak-search-elastic/src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexImporter.java b/oak-search-elastic/src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexImporter.java
index 7377cc1..433f50b 100644
--- a/oak-search-elastic/src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexImporter.java
+++ b/oak-search-elastic/src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexImporter.java
@@ -23,12 +23,15 @@ import org.apache.jackrabbit.oak.plugins.index.importer.IndexImporterProvider;
import org.apache.jackrabbit.oak.plugins.index.search.ReindexOperations;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
public class ElasticIndexImporter implements IndexImporterProvider {
+ private static final Logger LOG = LoggerFactory.getLogger(ElasticIndexCleaner.class);
public ElasticIndexImporter(){
}
@@ -36,6 +39,7 @@ public class ElasticIndexImporter implements IndexImporterProvider {
@Override
public void importIndex(NodeState root, NodeBuilder definitionBuilder, File indexDir) throws IOException, CommitFailedException {
// NOOP for elastic
+ LOG.info("No need to import data in case of elastic since this is a remote index. Exiting with NOOP");
}
@Override
diff --git a/oak-search/src/main/java/org/apache/jackrabbit/oak/plugins/index/search/ReindexOperations.java b/oak-search/src/main/java/org/apache/jackrabbit/oak/plugins/index/search/ReindexOperations.java
index 1e2f4d0..d74f4d2 100644
--- a/oak-search/src/main/java/org/apache/jackrabbit/oak/plugins/index/search/ReindexOperations.java
+++ b/oak-search/src/main/java/org/apache/jackrabbit/oak/plugins/index/search/ReindexOperations.java
@@ -19,6 +19,7 @@
package org.apache.jackrabbit.oak.plugins.index.search;
+import org.apache.jackrabbit.oak.plugins.index.importer.IndexImporter;
import org.apache.jackrabbit.oak.plugins.index.search.util.NodeStateCloner;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
@@ -63,6 +64,7 @@ public class ReindexOperations {
public IndexDefinition apply(boolean useStateFromBuilder) {
IndexFormatVersion version = IndexDefinition.determineVersionForFreshIndex(definitionBuilder);
definitionBuilder.setProperty(IndexDefinition.INDEX_VERSION, version.getVersion());
+ definitionBuilder.removeProperty(IndexImporter.INDEX_IMPORT_STATE_KEY);
//Avoid obtaining the latest NodeState from builder as that would force purge of current transient state
//as index definition does not get modified as part of IndexUpdate run in most case we rely on base state
@@ -70,6 +72,7 @@ public class ReindexOperations {
NodeState defnState = useStateFromBuilder ? definitionBuilder.getNodeState() : definitionBuilder.getBaseState();
if (storedIndexDefinitionEnabled) {
definitionBuilder.setChildNode(INDEX_DEFINITION_NODE, NodeStateCloner.cloneVisibleState(defnState));
+ definitionBuilder.setChildNode(INDEX_DEFINITION_NODE, NodeStateCloner.cloneVisibleState(defnState));
if (definitionBuilder.getChildNode(STATUS_NODE).exists()) {
definitionBuilder.getChildNode(STATUS_NODE).removeProperty(REINDEX_COMPLETION_TIMESTAMP);
LOG.info("{} property removed for index at {}", REINDEX_COMPLETION_TIMESTAMP, this.indexPath);
diff --git a/oak-search/src/test/java/org/apache/jackrabbit/oak/plugins/index/IndexImporterReindexTest.java b/oak-search/src/test/java/org/apache/jackrabbit/oak/plugins/index/IndexImporterReindexTest.java
new file mode 100644
index 0000000..148fbc9
--- /dev/null
+++ b/oak-search/src/test/java/org/apache/jackrabbit/oak/plugins/index/IndexImporterReindexTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+import org.apache.jackrabbit.oak.api.Tree;
+import org.apache.jackrabbit.oak.api.Type;
+import org.apache.jackrabbit.oak.plugins.index.importer.IndexImporter;
+import org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants;
+import org.apache.jackrabbit.oak.query.AbstractQueryTest;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Set;
+
+public abstract class IndexImporterReindexTest extends AbstractQueryTest {
+
+ protected IndexOptions indexOptions;
+ protected TestRepository repositoryOptionsUtil;
+
+ @Test
+ public void indexImporterFlagDeleteOnReindex() throws Exception {
+ Tree luceneIndex = createIndex("test-index", Collections.<String>emptySet());
+ Tree func = luceneIndex.addChild(FulltextIndexConstants.INDEX_RULES)
+ .addChild("nt:base")
+ .addChild(FulltextIndexConstants.PROP_NODE)
+ .addChild("lowerLocalName");
+ func.setProperty(FulltextIndexConstants.PROP_FUNCTION, "lower(localname())");
+
+ root.commit();
+ Assert.assertFalse(root.getTree("/oak:index/test-index").getProperty("reindex").getValue(Type.BOOLEAN));
+ root.getTree("/oak:index/test-index").setProperty(IndexImporter.INDEX_IMPORT_STATE_KEY, "test", Type.STRING);
+ root.getTree("/oak:index/test-index").setProperty("reindex", true, Type.BOOLEAN);
+ root.commit();
+ Assert.assertNull(root.getTree("/oak:index/test-index").getProperty(IndexImporter.INDEX_IMPORT_STATE_KEY));
+ }
+
+ protected Tree createIndex(String name, Set<String> propNames) {
+ Tree index = root.getTree("/");
+ return createIndex(index, name, propNames);
+ }
+
+ abstract protected Tree createIndex(Tree index, String name, Set<String> propNames);
+
+ abstract protected String getLoggerName();
+}