diff --git oak-api/src/main/java/org/apache/jackrabbit/oak/api/jmx/IndexStatsMBean.java oak-api/src/main/java/org/apache/jackrabbit/oak/api/jmx/IndexStatsMBean.java
index 6c6ce3f..cbda88f 100644
--- oak-api/src/main/java/org/apache/jackrabbit/oak/api/jmx/IndexStatsMBean.java
+++ oak-api/src/main/java/org/apache/jackrabbit/oak/api/jmx/IndexStatsMBean.java
@@ -104,6 +104,15 @@ public interface IndexStatsMBean {
long getUpdates();
/**
+ * Returns the number of which have been read so far. This value is
+ * kept until the next cycle begins.
+ *
+ * @return the number of node read from the current run cycle. This value is
+ * kept until the next cycle begins.
+ */
+ long getNodesReadCount();
+
+ /**
* Returns the current reference checkpoint used by the async indexer
*
* @return the reference checkpoint
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 6139a9e..6d79d79 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
@@ -236,8 +236,12 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
*
* @see OAK-1292
*/
- protected static class AsyncUpdateCallback implements IndexUpdateCallback {
-
+ protected static class AsyncUpdateCallback implements IndexUpdateCallback, NodeTraversalCallback {
+ /**
+ * Interval in terms of number of nodes traversed after which
+ * time would be checked for lease expiry
+ */
+ public static final int LEASE_CHECK_INTERVAL = 100;
private final NodeStore store;
/** The base checkpoint */
@@ -324,7 +328,7 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
mergeWithConcurrencyCheck(store, validatorProviders, builder, checkpoint, lease, name);
// reset updates counter
- indexStats.resetUpdates();
+ indexStats.reset();
}
private void updateTempCheckpoints(NodeBuilder async,
@@ -367,12 +371,23 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
@Override
public void indexUpdate() throws CommitFailedException {
- if (forcedStop.get()){
- forcedStop.set(false);
- throw INTERRUPTED;
- }
+ checkIfStopped();
+ indexStats.incUpdates();
+ }
+
+ public void setCheckpoint(String checkpoint) {
+ this.checkpoint = checkpoint;
+ }
+
+ public void setValidatorProviders(List validatorProviders) {
+ this.validatorProviders = checkNotNull(validatorProviders);
+ }
+
+ @Override
+ public void traversedNode() throws CommitFailedException{
+ checkIfStopped();
- if (indexStats.incUpdates() % 100 == 0 && isLeaseCheckEnabled(leaseTimeOut)) {
+ if (indexStats.incTraversal() % LEASE_CHECK_INTERVAL == 0 && isLeaseCheckEnabled(leaseTimeOut)) {
long now = System.currentTimeMillis();
if (now + leaseTimeOut > lease) {
long newLease = now + 2 * leaseTimeOut;
@@ -384,12 +399,11 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
}
}
- public void setCheckpoint(String checkpoint) {
- this.checkpoint = checkpoint;
- }
-
- public void setValidatorProviders(List validatorProviders) {
- this.validatorProviders = checkNotNull(validatorProviders);
+ private void checkIfStopped() throws CommitFailedException {
+ if (forcedStop.get()){
+ forcedStop.set(false);
+ throw INTERRUPTED;
+ }
}
}
@@ -888,6 +902,7 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
private volatile boolean isPaused;
private volatile long updates;
+ private volatile long nodesRead;
private final Stopwatch watch = Stopwatch.createUnstarted();
private final ExecutionStats execStats;
@@ -1034,13 +1049,17 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
return this.isPaused;
}
- void resetUpdates() {
+ void reset() {
this.updates = 0;
+ this.nodesRead = 0;
}
long incUpdates() {
- updates++;
- return updates;
+ return ++updates;
+ }
+
+ long incTraversal() {
+ return ++nodesRead;
}
@Override
@@ -1048,6 +1067,11 @@ public class AsyncIndexUpdate implements Runnable, Closeable {
return updates;
}
+ @Override
+ public long getNodesReadCount(){
+ return nodesRead;
+ }
+
void setReferenceCheckpoint(String checkpoint) {
this.referenceCp = checkpoint;
}
diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUpdate.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUpdate.java
index 61cffe0..025315e 100644
--- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUpdate.java
+++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUpdate.java
@@ -497,6 +497,7 @@ public class IndexUpdate implements Editor {
* Callback for the update events of the indexing job
*/
final IndexUpdateCallback updateCallback;
+ final NodeTraversalCallback traversalCallback;
final Set reindexedIndexes = Sets.newHashSet();
final Map callbacks = Maps.newHashMap();
final CorruptIndexHandler corruptIndexHandler;
@@ -512,6 +513,7 @@ public class IndexUpdate implements Editor {
this.updateCallback = checkNotNull(updateCallback);
this.commitInfo = commitInfo;
this.corruptIndexHandler = corruptIndexHandler;
+ this.traversalCallback = updateCallback instanceof NodeTraversalCallback ? (NodeTraversalCallback)updateCallback : NodeTraversalCallback.NOOP;
}
public IndexUpdateCallback newCallback(String indexPath, boolean reindex) {
@@ -570,8 +572,9 @@ public class IndexUpdate implements Editor {
return !reindexedIndexes.isEmpty();
}
- public void nodeRead(String name){
+ public void nodeRead(String name) throws CommitFailedException {
changedNodeCount++;
+ traversalCallback.traversedNode();
}
public void propertyChanged(String name){
diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/NodeTraversalCallback.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/NodeTraversalCallback.java
new file mode 100644
index 0000000..412dea1
--- /dev/null
+++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/NodeTraversalCallback.java
@@ -0,0 +1,38 @@
+/*
+ * 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.CommitFailedException;
+
+/**
+ * Callback which invoked for any changed node read by IndexUpdate
+ * as part of diff traversal
+ */
+interface NodeTraversalCallback{
+ NodeTraversalCallback NOOP = new NodeTraversalCallback() {
+ @Override
+ public void traversedNode() throws CommitFailedException {
+
+ }
+ };
+
+
+ void traversedNode() throws CommitFailedException;
+}
diff --git oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java
index 91fc60e..52f07fa 100644
--- oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java
+++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java
@@ -1773,6 +1773,47 @@ public class AsyncIndexUpdateTest {
}
}
+ @Test
+ public void traversalCount() throws Exception{
+ MemoryNodeStore store = new MemoryNodeStore();
+ PropertyIndexEditorProvider provider = new PropertyIndexEditorProvider();
+
+ NodeBuilder builder = store.getRoot().builder();
+ createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME),
+ "rootIndex", true, false, ImmutableSet.of("foo"), null)
+ .setProperty(ASYNC_PROPERTY_NAME, "async");
+ builder.child("testRoot").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider);
+ async.run();
+
+ //Get rid of changes in index nodes i.e. /oak:index/rootIndex
+ async.run();
+
+ //Do a run without any index property change
+ builder = store.getRoot().builder();
+ builder.child("a").child("b");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ async.run();
+
+ AsyncIndexStats stats = async.getIndexStats();
+ assertEquals(2, stats.getNodesReadCount());
+ assertEquals(0, stats.getUpdates());
+
+ //Do a run with a index property change
+ builder = store.getRoot().builder();
+ builder.child("a").child("b").setProperty("foo", "bar");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ async.run();
+
+ stats = async.getIndexStats();
+ assertEquals(2, stats.getNodesReadCount());
+ assertEquals(1, stats.getUpdates());
+ }
+
private static class TestIndexEditorProvider extends PropertyIndexEditorProvider {
private String indexPathToFail;
@Override
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 b656155..7641762 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
@@ -24,9 +24,11 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertFalse;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.collect.Sets;
import org.apache.jackrabbit.oak.OakBaseTest;
import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.fixture.NodeStoreFixture;
@@ -396,8 +398,45 @@ public class AsyncIndexUpdateLeaseTest extends OakBaseTest {
.hasProperty(AsyncIndexUpdate.leasify(name)));
}
+ @Test
+ public void testLeaseUpdateAndNumberOfChanges() throws Exception {
+ final long lease = 50;
+ testContent(store, AsyncUpdateCallback.LEASE_CHECK_INTERVAL / 2);
+ Set leaseTimes = Sets.newHashSet();
+ final IndexStatusListener l1 = new IndexStatusListener() {
+ @Override
+ protected void postIndexUpdate() {
+ leaseTimes.add(getLeaseValue());
+ if (!executed.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(lease * 3);
+ } catch (InterruptedException e) {
+ //
+ }
+ }
+ executed.set(true);
+ }
+ };
+ assertRunOk(new SpecialAsyncIndexUpdate(name, store, provider, l1)
+ .setLeaseTimeOut(lease));
+
+ assertEquals(1, leaseTimes.size());
+
+ leaseTimes.clear();
+
+ //Run with changes more than threshold and then lease should change more than once
+ testContent(store, AsyncUpdateCallback.LEASE_CHECK_INTERVAL * 2);
+ assertRunOk(new SpecialAsyncIndexUpdate(name, store, provider, l1)
+ .setLeaseTimeOut(lease));
+ assertTrue(leaseTimes.size() > 1);
+ }
+
// -------------------------------------------------------------------
+ private long getLeaseValue() {
+ return store.getRoot().getChildNode(":async").getLong( AsyncIndexUpdate.leasify(name));
+ }
+
private static String getReferenceCp(NodeStore store, String name) {
return store.getRoot().getChildNode(AsyncIndexUpdate.ASYNC)
.getString(name);
@@ -429,6 +468,16 @@ public class AsyncIndexUpdateLeaseTest extends OakBaseTest {
store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
}
+ private static void testContent(NodeStore store, int numOfNodes) throws Exception {
+ NodeBuilder builder = store.getRoot().builder();
+ for (int i = 0; i < numOfNodes; i++) {
+ builder.child("testRoot"+i).setProperty("foo",
+ "abc " + System.currentTimeMillis());
+ }
+
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ }
+
private static class SpecialAsyncIndexUpdate extends AsyncIndexUpdate {
private final IndexStatusListener listener;