diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java index c923b8f..4aedf8e 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java @@ -146,6 +146,8 @@ public class AsyncIndexUpdate implements Runnable { private final IndexTaskSpliter taskSplitter = new IndexTaskSpliter(); + private long leaseTimeOut; + private IndexMBeanRegistration mbeanRegistration; public AsyncIndexUpdate(@Nonnull String name, @Nonnull NodeStore store, @@ -156,6 +158,7 @@ public class AsyncIndexUpdate implements Runnable { this.store = checkNotNull(store); this.provider = checkNotNull(provider); this.switchOnSync = switchOnSync; + this.leaseTimeOut = ASYNC_TIMEOUT; } public AsyncIndexUpdate(@Nonnull String name, @Nonnull NodeStore store, @@ -185,7 +188,7 @@ public class AsyncIndexUpdate implements Runnable { throws CommitFailedException { long now = System.currentTimeMillis(); this.checkpoint = checkpoint; - this.lease = now + 2 * ASYNC_TIMEOUT; + this.lease = now + 2 * leaseTimeOut; this.leaseName = name + "-lease"; NodeState root = store.getRoot(); @@ -245,8 +248,8 @@ public class AsyncIndexUpdate implements Runnable { if (updates % 100 == 0) { indexStats.setUpdates(this.updates); long now = System.currentTimeMillis(); - if (now + ASYNC_TIMEOUT > lease) { - long newLease = now + 2 * ASYNC_TIMEOUT; + if (now + leaseTimeOut > lease) { + long newLease = now + 2 * leaseTimeOut; NodeBuilder builder = store.getRoot().builder(); builder.child(ASYNC).setProperty(leaseName, newLease); mergeWithConcurrencyCheck(builder, checkpoint, lease); @@ -379,6 +382,10 @@ public class AsyncIndexUpdate implements Runnable { } } + public void setLeaseTimeOut(long leaseTimeOut, TimeUnit timeUnit) { + this.leaseTimeOut = timeUnit.toMillis(leaseTimeOut); + } + private boolean updateIndex( NodeState before, String beforeCheckpoint, NodeState after, String afterCheckpoint, String afterTime) diff --git oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateClusterTest.java oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateClusterTest.java new file mode 100644 index 0000000..e55fba3 --- /dev/null +++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateClusterTest.java @@ -0,0 +1,309 @@ +/* + * 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 java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import javax.annotation.Nonnull; + +import com.google.common.collect.ImmutableSet; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.api.PropertyState; +import org.apache.jackrabbit.oak.plugins.document.DocumentMK; +import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore; +import org.apache.jackrabbit.oak.plugins.document.memory.MemoryDocumentStore; +import org.apache.jackrabbit.oak.plugins.index.property.PropertyIndexEditorProvider; +import org.apache.jackrabbit.oak.spi.blob.MemoryBlobStore; +import org.apache.jackrabbit.oak.spi.commit.CommitInfo; +import org.apache.jackrabbit.oak.spi.commit.Editor; +import org.apache.jackrabbit.oak.spi.commit.EmptyHook; +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.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.ASYNC_PROPERTY_NAME; +import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.INDEX_DEFINITIONS_NAME; +import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.MISSING_NODE; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class AsyncIndexUpdateClusterTest { + private final Logger log = LoggerFactory.getLogger(getClass()); + private DocumentNodeStore ns1; + private DocumentNodeStore ns2; + + @Before + public void before() throws Exception { + MemoryDocumentStore ds = new MemoryDocumentStore(); + MemoryBlobStore bs = new MemoryBlobStore(); + DocumentMK.Builder builder; + + builder = new DocumentMK.Builder(); + builder.setDocumentStore(ds).setBlobStore(bs).setAsyncDelay(0); + ns1 = builder.setClusterId(1).getNodeStore(); + builder = new DocumentMK.Builder(); + builder.setDocumentStore(ds).setBlobStore(bs).setAsyncDelay(0); + ns2 = builder.setClusterId(2).getNodeStore(); + } + + @After + public void after() { + ns1.dispose(); + ns2.dispose(); + } + + @Test + public void missingCheckpointDueToEventualConsistency() throws Exception { + IndexStatusListener l1 = new IndexStatusListener(); + + IndexStatusListener l2 = new IndexStatusListener(); + + AsyncIndexUpdate async1 = createAsync(ns1, l1); + AsyncIndexUpdate async2 = createAsync(ns2, l2); + + // Set timeout to 0 + async1.setLeaseTimeOut(0, TimeUnit.MILLISECONDS); + async2.setLeaseTimeOut(0, TimeUnit.MILLISECONDS); + + // Phase 1 - Base setup - Index definition creation and + // performing initial indexing + // Create index definition on NS1 + NodeBuilder b1 = ns1.getRoot().builder(); + createIndexDefinition(b1); + ns1.merge(b1, EmptyHook.INSTANCE, CommitInfo.EMPTY); + + //Create some node to be indexed + createTestNode(ns1, "t1"); + + //Trigger indexing on NS1 + async1.run(); + + //Make the changes visible to other cluster nodes + ns1.runBackgroundOperations(); + ns2.runBackgroundOperations(); + + log.info("===== Phase 1 - Created base setup"); + + // Phase 2 - Create scenario. Start indexing on NS2 and make it + // wait in indexing phase + createTestNode(ns2, "t2"); + l2.waitOn("t2"); + Thread asyncThread_N2 = new Thread(async2); + asyncThread_N2.start(); + + // By this time the tmp checkpoint array would be + // updated to have the new checkpoint as an entry + l2.waitForIndexerToArrive(); + + // Publish the changes so far on N2 i.e. it would publish + // tmp checkpoint array + ns2.runBackgroundOperations(); + + // Make changes visible on N1 + ns1.runBackgroundOperations(); + + log.info("===== Phase 2 - async2 now waiting for signal"); + // Phase 3 - Now start async indexer of N1 and make it also + // wait at indexing phase. + createTestNode(ns1, "t3"); + + l1.waitOn("t3"); + Thread asyncThread_N1 = new Thread(async1); + asyncThread_N1.start(); + + // By this time it would have released the checkpoint + // created by async2 as it was part of tmp + l1.waitForIndexerToArrive(); + + log.info("===== Phase 3 - async1 now waiting for signal"); + //Phase 4 - Now let both async complete. async2 completes first and + // thus prevent async1 from completion + l2.resume(); + asyncThread_N2.join(); + + l1.resume(); + asyncThread_N1.join(); + assertTrue(async1.isFailing()); + + ns2.runBackgroundOperations(); + ns1.runBackgroundOperations(); + + log.info("===== Phase 4 - Both async allowed to complete"); + // Phase 5 - Now run another round of indexing on NS2. + // For now this would fail because checkpoint would be missing + createTestNode(ns2, "t4"); + l2.resetIndexingStatus(); + async2.run(); + + log.info("===== Phase 5 - Done with next cycle on async2"); + assertFalse("Reindexing should not have happened", l2.reindexed); + } + + private static AsyncIndexUpdate createAsync(NodeStore ns, IndexStatusListener l) { + IndexEditorProvider p = new TestEditorProvider(new PropertyIndexEditorProvider(), l); + return new AsyncIndexUpdate("async", ns, p); + } + + private static void createIndexDefinition(NodeBuilder builder) { + IndexUtils.createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), + "rootIndex", true, false, ImmutableSet.of("foo"), null) + .setProperty(ASYNC_PROPERTY_NAME, "async"); + } + + private static void createTestNode(NodeStore ns, String nodeName) throws CommitFailedException { + NodeBuilder b = ns.getRoot().builder(); + b.child(nodeName).setProperty("foo", "abc"); + ns.merge(b, EmptyHook.INSTANCE, CommitInfo.EMPTY); + } + + + private static class IndexStatusListener { + boolean reindexed; + private final CountDownLatch indexer = new CountDownLatch(1); + private final CountDownLatch tester = new CountDownLatch(1); + private volatile String nodeNameToWaitOn; + + public void reindexingDone() { + reindexed = true; + } + + public void resetIndexingStatus() { + reindexed = false; + } + + public void waitIfRequired(String nodeName) { + if (nodeName.equals(nodeNameToWaitOn)) { + try { + tester.countDown(); + indexer.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + public void waitOn(String nodeName) { + this.nodeNameToWaitOn = nodeName; + } + + public void waitForIndexerToArrive() throws InterruptedException { + tester.await(); + } + + public void resume() throws InterruptedException { + indexer.countDown(); + } + } + + private static class TestEditorProvider implements IndexEditorProvider { + private final IndexEditorProvider delegate; + private final IndexStatusListener listener; + + private TestEditorProvider(IndexEditorProvider delegate, IndexStatusListener listener) { + this.delegate = delegate; + this.listener = listener; + } + + @Override + public Editor getIndexEditor(@Nonnull String type, @Nonnull NodeBuilder definition, + @Nonnull NodeState root, @Nonnull IndexUpdateCallback callback) throws CommitFailedException { + Editor e = delegate.getIndexEditor(type, definition, root, callback); + if (e != null) { + e = new TestEditor(e, listener); + } + return e; + } + } + + private static class TestEditor implements Editor { + private final Editor editor; + private final TestEditor parent; + private final IndexStatusListener listener; + + TestEditor(Editor editor, IndexStatusListener listener) { + this(editor, listener, null); + } + + TestEditor(Editor editor, IndexStatusListener listener, TestEditor parent) { + this.editor = editor; + this.listener = listener; + this.parent = parent; + } + + + @Override + public void enter(NodeState before, NodeState after) throws CommitFailedException { + if (MISSING_NODE == before && parent == null) { + listener.reindexingDone(); + } + editor.enter(before, after); + } + + @Override + public void leave(NodeState before, NodeState after) throws CommitFailedException { + editor.leave(before, after); + } + + @Override + public void propertyAdded(PropertyState after) throws CommitFailedException { + editor.propertyAdded(after); + } + + @Override + public void propertyChanged(PropertyState before, PropertyState after) throws CommitFailedException { + editor.propertyChanged(before, after); + } + + @Override + public void propertyDeleted(PropertyState before) throws CommitFailedException { + editor.propertyDeleted(before); + } + + @Override + public Editor childNodeAdded(String name, NodeState after) throws CommitFailedException { + listener.waitIfRequired(name); + return createChildEditor(editor.childNodeAdded(name, after), name); + } + + @Override + public Editor childNodeChanged(String name, NodeState before, NodeState after) throws CommitFailedException { + return createChildEditor(editor.childNodeChanged(name, before, after), name); + } + + @Override + public Editor childNodeDeleted(String name, NodeState before) throws CommitFailedException { + return createChildEditor(editor.childNodeDeleted(name, before), name); + } + + private TestEditor createChildEditor(Editor editor, String name) { + if (editor == null) { + return null; + } else { + return new TestEditor(editor, listener, this); + } + } + } +} diff --git oak-core/src/test/resources/logback-test.xml oak-core/src/test/resources/logback-test.xml index 7a9abc4..fa33469 100644 --- oak-core/src/test/resources/logback-test.xml +++ oak-core/src/test/resources/logback-test.xml @@ -24,6 +24,7 @@ target/unit-tests.log + false %date{HH:mm:ss.SSS} %-5level %-40([%thread] %F:%L) %msg%n @@ -36,4 +37,6 @@ + +