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 d973794..af9439b 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 @@ -266,6 +266,10 @@ public class AsyncIndexUpdate implements Runnable, Closeable { } protected void initLease() throws CommitFailedException { + if (!isLeaseCheckEnabled()){ + return; + } + if (hasLease) { return; } @@ -343,7 +347,7 @@ public class AsyncIndexUpdate implements Runnable, Closeable { throw INTERRUPTED; } - if (indexStats.incUpdates() % 100 == 0) { + if (indexStats.incUpdates() % 100 == 0 && isLeaseCheckEnabled()) { long now = System.currentTimeMillis(); if (now + leaseTimeOut > lease) { long newLease = now + 2 * leaseTimeOut; @@ -355,6 +359,10 @@ public class AsyncIndexUpdate implements Runnable, Closeable { } } + private boolean isLeaseCheckEnabled(){ + return leaseTimeOut > 0; + } + public void setCheckpoint(String checkpoint) { this.checkpoint = checkpoint; } diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerService.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerService.java index cedcb43..f6d9a2a 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerService.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerService.java @@ -36,6 +36,7 @@ import org.apache.jackrabbit.oak.commons.PropertiesUtil; import org.apache.jackrabbit.oak.osgi.OsgiWhiteboard; import org.apache.jackrabbit.oak.plugins.observation.ChangeCollectorProvider; import org.apache.jackrabbit.oak.spi.commit.ValidatorProvider; +import org.apache.jackrabbit.oak.spi.state.Clusterable; import org.apache.jackrabbit.oak.spi.state.NodeStore; import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard; import org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardIndexEditorProvider; @@ -97,6 +98,11 @@ public class AsyncIndexerService { long leaseTimeOutMin = PropertiesUtil.toInteger(config.get(PROP_LEASE_TIME_OUT), PROP_LEASE_TIMEOUT_DEFAULT); + if (!(nodeStore instanceof Clusterable)){ + leaseTimeOutMin = 0; + log.info("Detected non clusterable setup. Lease checking would be disabled for async indexing"); + } + for (AsyncConfig c : asyncIndexerConfig) { AsyncIndexUpdate task = new AsyncIndexUpdate(c.name, nodeStore, indexEditorProvider, statisticsProvider, false); diff --git oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerServiceTest.java oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerServiceTest.java index 1773983..00499a0 100644 --- oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerServiceTest.java +++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexerServiceTest.java @@ -23,6 +23,8 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import javax.annotation.Nonnull; + import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import org.apache.jackrabbit.oak.plugins.index.AsyncIndexUpdateTest.CommitInfoCollector; @@ -35,6 +37,7 @@ import org.apache.jackrabbit.oak.spi.commit.CommitContext; import org.apache.jackrabbit.oak.spi.commit.CommitInfo; import org.apache.jackrabbit.oak.spi.commit.EmptyHook; import org.apache.jackrabbit.oak.spi.commit.ValidatorProvider; +import org.apache.jackrabbit.oak.spi.state.Clusterable; import org.apache.jackrabbit.oak.spi.state.NodeBuilder; import org.apache.jackrabbit.oak.spi.state.NodeStore; import org.apache.jackrabbit.oak.stats.StatisticsProvider; @@ -53,19 +56,13 @@ public class AsyncIndexerServiceTest { @Rule public final OsgiContext context = new OsgiContext(); - private MemoryNodeStore nodeStore = new MemoryNodeStore(); + private MemoryNodeStore nodeStore = new FakeClusterableMemoryNodeStore(); private AsyncIndexerService service = new AsyncIndexerService(); - @Before - public void setUp() { - context.registerService(StatisticsProvider.class, StatisticsProvider.NOOP); - context.registerService(NodeStore.class, nodeStore); - context.registerService(ValidatorProvider.class, new ChangeCollectorProvider()); - MockOsgi.injectServices(service, context.bundleContext()); - } @Test public void asyncReg() throws Exception{ + injectDefaultServices(); Map config = ImmutableMap.of( "asyncConfigs", new String[] {"async:5"} ); @@ -79,6 +76,7 @@ public class AsyncIndexerServiceTest { @Test public void leaseTimeout() throws Exception{ + injectDefaultServices(); Map config = ImmutableMap.of( "asyncConfigs", new String[] {"async:5"}, "leaseTimeOutMinutes" , "20" @@ -90,6 +88,7 @@ public class AsyncIndexerServiceTest { @Test public void changeCollectionEnabled() throws Exception{ + injectDefaultServices(); Map config = ImmutableMap.of( "asyncConfigs", new String[] {"async:5"} ); @@ -117,8 +116,18 @@ public class AsyncIndexerServiceTest { assertNotNull(changeSet); } - private AsyncIndexUpdate getIndexUpdate(String name) { - return (AsyncIndexUpdate) context.getServices(Runnable.class, "(oak.async="+name+")")[0]; + @Test + public void nonClusterableNodeStoreAndLeaseTimeout() throws Exception{ + nodeStore = new MemoryNodeStore(); + injectDefaultServices(); + + Map config = ImmutableMap.of( + "asyncConfigs", new String[] {"async:5"}, + "leaseTimeOutMinutes" , "20" + ); + MockOsgi.activate(service, context.bundleContext(), config); + AsyncIndexUpdate indexUpdate = getIndexUpdate("async"); + assertEquals(0, indexUpdate.getLeaseTimeOut()); } @Test @@ -133,4 +142,23 @@ public class AsyncIndexerServiceTest { assertEquals("foo", configs.get(1).name); assertEquals(23, configs.get(1).timeIntervalInSecs); } + + private void injectDefaultServices() { + context.registerService(StatisticsProvider.class, StatisticsProvider.NOOP); + context.registerService(NodeStore.class, nodeStore); + context.registerService(ValidatorProvider.class, new ChangeCollectorProvider()); + MockOsgi.injectServices(service, context.bundleContext()); + } + + private AsyncIndexUpdate getIndexUpdate(String name) { + return (AsyncIndexUpdate) context.getServices(Runnable.class, "(oak.async="+name+")")[0]; + } + + private static class FakeClusterableMemoryNodeStore extends MemoryNodeStore implements Clusterable { + @Nonnull + @Override + public String getInstanceId() { + return "foo"; + } + } } \ No newline at end of file diff --git oak-it/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateLeaseTest.java oak-it/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateLeaseTest.java index df72e53..db8bf66 100644 --- oak-it/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateLeaseTest.java +++ oak-it/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateLeaseTest.java @@ -349,6 +349,22 @@ public class AsyncIndexUpdateLeaseTest extends OakBaseTest { .setLeaseTimeOut(lease)); } + + @Test + public void testLeaseDisabled() throws Exception { + // take care of initial reindex before + AsyncIndexUpdate async = new AsyncIndexUpdate(name, store, provider).setLeaseTimeOut(0); + async.run(); + + testContent(store); + assertRunOk(async); + + testContent(store); + assertRunOk(async); + + executed.set(true); + } + // ------------------------------------------------------------------- private static String getReferenceCp(NodeStore store, String name) {