nfsantos commented on code in PR #1522:
URL: https://github.com/apache/jackrabbit-oak/pull/1522#discussion_r1638287794
##########
oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java:
##########
@@ -1242,6 +1242,54 @@ public String getReferenceCheckpoint() {
return referenceCp;
}
+ @Override
+ public String forceIndexLaneCatchup(String confirmMessage) throws
CommitFailedException {
+
+ if (!"CONFIRM".equals(confirmMessage)) {
+ String msg = "Please confirm that you want to force the lane
catchup by passing 'CONFIRM' as argument";
+ log.warn(msg);
+ return msg;
+ }
+
+ if (!this.isFailing()) {
+ String msg = "The lane is not failing. This operation should
only be performed if the lane is failing, it should first be allowed to catchup
on its own.";
+ log.warn(msg);
+ return msg;
+ }
+
+ log.info("Running a forced catchup for indexing lane [{}]", name);
+ // First we need to abort and pause the running indexing task
+ this.abortAndPause();
+ log.info("Aborted and paused async indexing for lane [{}]", name);
+ // Release lease for the paused lane
+ this.releaseLeaseForPausedLane();
+ log.info("Released lease for paused lane [{}]", name);
+ String newReferenceCheckpoint = store.checkpoint(lifetime,
ImmutableMap.of(
+ "creator", AsyncIndexUpdate.class.getSimpleName(),
+ "created", now(),
+ "thread", Thread.currentThread().getName(),
+ "name", name + "-forceModified"));
+ String existingReferenceCheckpoint = this.referenceCp;
+ log.info("Modifying the referred checkpoint for lane [{}] from {}
to {}." +
+ " This means that any content modifications b/w these
checkpoints will not reflect in the indexes on this lane." +
+ " Reindexing would be needed to get this content
indexed.", name, existingReferenceCheckpoint, newReferenceCheckpoint);
Review Comment:
```suggestion
" Reindexing is needed to get this content indexed.",
name, existingReferenceCheckpoint, newReferenceCheckpoint);
```
##########
oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java:
##########
@@ -1242,6 +1242,54 @@ public String getReferenceCheckpoint() {
return referenceCp;
}
+ @Override
+ public String forceIndexLaneCatchup(String confirmMessage) throws
CommitFailedException {
+
+ if (!"CONFIRM".equals(confirmMessage)) {
+ String msg = "Please confirm that you want to force the lane
catchup by passing 'CONFIRM' as argument";
+ log.warn(msg);
+ return msg;
+ }
+
+ if (!this.isFailing()) {
+ String msg = "The lane is not failing. This operation should
only be performed if the lane is failing, it should first be allowed to catchup
on its own.";
+ log.warn(msg);
+ return msg;
+ }
+
+ log.info("Running a forced catchup for indexing lane [{}]", name);
+ // First we need to abort and pause the running indexing task
+ this.abortAndPause();
+ log.info("Aborted and paused async indexing for lane [{}]", name);
+ // Release lease for the paused lane
+ this.releaseLeaseForPausedLane();
+ log.info("Released lease for paused lane [{}]", name);
+ String newReferenceCheckpoint = store.checkpoint(lifetime,
ImmutableMap.of(
+ "creator", AsyncIndexUpdate.class.getSimpleName(),
+ "created", now(),
+ "thread", Thread.currentThread().getName(),
+ "name", name + "-forceModified"));
+ String existingReferenceCheckpoint = this.referenceCp;
+ log.info("Modifying the referred checkpoint for lane [{}] from {}
to {}." +
+ " This means that any content modifications b/w these
checkpoints will not reflect in the indexes on this lane." +
+ " Reindexing would be needed to get this content
indexed.", name, existingReferenceCheckpoint, newReferenceCheckpoint);
+ NodeBuilder builder = store.getRoot().builder();
+ builder.child(ASYNC).setProperty(name, newReferenceCheckpoint);
+ this.referenceCp = newReferenceCheckpoint;
+ // Remove the existing reference checkpoint
+ if (store.release(existingReferenceCheckpoint)) {
+ log.info("Old reference checkpoint {} removed or didn't
exist", existingReferenceCheckpoint);
+ } else {
+ log.warn("Unable to remove old reference checkpoint {}. This
can result in orphaned checkpoints and would need to be removed manually.",
existingReferenceCheckpoint);
+ }
+ mergeWithConcurrencyCheck(store, validatorProviders, builder,
null, null, name);
+
+ // Resume the paused lane;
+ this.resume();
+ log.info("Resumed async indexing for lane [{}]", name);
+ return ("Lane successfully forced to catchup. New reference
checkpoint is " + newReferenceCheckpoint + " . Please make sure to perform
reindexing to get the diff content indexed.");
Review Comment:
I think the parenthesis are unnecessary.
##########
oak-api/src/main/java/org/apache/jackrabbit/oak/api/jmx/IndexStatsMBean.java:
##########
@@ -127,6 +128,15 @@ public interface IndexStatsMBean {
*/
String getReferenceCheckpoint();
+ @Description("Force update the indexing lane to a latest checkpoint. This
will abort and pause the running lane, release its lease and set the reference
checkpoint to a latest one." +
+ "Any content changes b/w the old reference checkpoint and the new
one will be not be indexed and a reindexing would be required." +
Review Comment:
I would prefer to not use abbreviations in user facing documentation, so use
`between` instead of `b/w`. I'm not even sure that `b/w` is meant to mean
`between`, I'm just guessing.
##########
oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java:
##########
@@ -589,6 +589,83 @@ public void run() {
assertNoConflictMarker(builder);
}
+ @Test
+ public void testForceUpdateAsyncLane() throws CommitFailedException {
+ NodeStore store = new MemoryNodeStore();
+ IndexEditorProvider provider = new PropertyIndexEditorProvider();
+
+ NodeBuilder builder = store.getRoot().builder();
+ createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME),
+ "rootIndex", true, false, ImmutableSet.of("foo"), null)
Review Comment:
Use JDK Set instead of Guava.
##########
oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java:
##########
@@ -589,6 +589,83 @@ public void run() {
assertNoConflictMarker(builder);
}
+ @Test
+ public void testForceUpdateAsyncLane() throws CommitFailedException {
+ NodeStore store = new MemoryNodeStore();
+ IndexEditorProvider 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");
+
+ // merge it back in
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ AsyncIndexUpdate async = new AsyncIndexUpdate("async", store,
provider);
+ async.run();
+ NodeState root = store.getRoot();
+ builder = root.builder();
+ builder.child("testRoot1").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ // Run force index catchup with an incorrect confirm message
+ // This will skip the operation and testRoot1 should be indexed in the
next async run.
+ async.getIndexStats().forceIndexLaneCatchup("ok");
+ async.run();
+ root = store.getRoot();
+
+ // first check that the index content nodes exist
+ checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex",
+ INDEX_CONTENT_NODE_NAME);
+ assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode(
+ ":conflict"));
+ PropertyIndexLookup lookup = new PropertyIndexLookup(root);
+ assertEquals(ImmutableSet.of("testRoot", "testRoot1"), find(lookup,
"foo", "abc"));
+
+ // Run force index catchup with correct confirm message
+ // But the async lane is NOT failing
+ // Due to this the force update should be skipped and the
+ // new node testRoot2 will be indexed.
+ builder.child("testRoot2").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ async.getIndexStats().forceIndexLaneCatchup("CONFIRM");
+ async.run();
+ root = store.getRoot();
+
+ checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex",
+ INDEX_CONTENT_NODE_NAME);
+ assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode(
+ ":conflict"));
+ lookup = new PropertyIndexLookup(root);
+ assertEquals(ImmutableSet.of("testRoot", "testRoot1", "testRoot2"),
find(lookup, "foo", "abc"));
Review Comment:
Guava vs JDK.
##########
oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java:
##########
@@ -1242,6 +1242,54 @@ public String getReferenceCheckpoint() {
return referenceCp;
}
+ @Override
+ public String forceIndexLaneCatchup(String confirmMessage) throws
CommitFailedException {
+
+ if (!"CONFIRM".equals(confirmMessage)) {
+ String msg = "Please confirm that you want to force the lane
catchup by passing 'CONFIRM' as argument";
+ log.warn(msg);
+ return msg;
+ }
+
+ if (!this.isFailing()) {
+ String msg = "The lane is not failing. This operation should
only be performed if the lane is failing, it should first be allowed to catchup
on its own.";
+ log.warn(msg);
+ return msg;
+ }
+
+ log.info("Running a forced catchup for indexing lane [{}]", name);
+ // First we need to abort and pause the running indexing task
+ this.abortAndPause();
+ log.info("Aborted and paused async indexing for lane [{}]", name);
+ // Release lease for the paused lane
+ this.releaseLeaseForPausedLane();
+ log.info("Released lease for paused lane [{}]", name);
+ String newReferenceCheckpoint = store.checkpoint(lifetime,
ImmutableMap.of(
+ "creator", AsyncIndexUpdate.class.getSimpleName(),
+ "created", now(),
+ "thread", Thread.currentThread().getName(),
+ "name", name + "-forceModified"));
+ String existingReferenceCheckpoint = this.referenceCp;
+ log.info("Modifying the referred checkpoint for lane [{}] from {}
to {}." +
+ " This means that any content modifications b/w these
checkpoints will not reflect in the indexes on this lane." +
Review Comment:
```suggestion
" This means that any content modifications between
these checkpoints will not reflect in the indexes on this lane." +
```
##########
oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java:
##########
@@ -1242,6 +1242,54 @@ public String getReferenceCheckpoint() {
return referenceCp;
}
+ @Override
+ public String forceIndexLaneCatchup(String confirmMessage) throws
CommitFailedException {
+
+ if (!"CONFIRM".equals(confirmMessage)) {
+ String msg = "Please confirm that you want to force the lane
catchup by passing 'CONFIRM' as argument";
+ log.warn(msg);
+ return msg;
+ }
+
+ if (!this.isFailing()) {
+ String msg = "The lane is not failing. This operation should
only be performed if the lane is failing, it should first be allowed to catchup
on its own.";
+ log.warn(msg);
+ return msg;
+ }
+
+ log.info("Running a forced catchup for indexing lane [{}]", name);
Review Comment:
Just a minor grammar detail. `catchup` is incorrect, the word does not
exist. Either
- catch up -> verb
- catch-up -> adjective/noum
So here it should be catch-up. There are other incorrect uses of `catchup`
in this PR. It's not too important, but as this user facing documentation I
think we should try to use correct grammar.
##########
oak-api/src/main/java/org/apache/jackrabbit/oak/api/jmx/IndexStatsMBean.java:
##########
@@ -127,6 +128,15 @@ public interface IndexStatsMBean {
*/
String getReferenceCheckpoint();
+ @Description("Force update the indexing lane to a latest checkpoint. This
will abort and pause the running lane, release its lease and set the reference
checkpoint to a latest one." +
Review Comment:
Can you be more specific of what latest mean? Exact values if possible.
##########
oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdate.java:
##########
@@ -1242,6 +1242,54 @@ public String getReferenceCheckpoint() {
return referenceCp;
}
+ @Override
+ public String forceIndexLaneCatchup(String confirmMessage) throws
CommitFailedException {
+
+ if (!"CONFIRM".equals(confirmMessage)) {
+ String msg = "Please confirm that you want to force the lane
catchup by passing 'CONFIRM' as argument";
+ log.warn(msg);
+ return msg;
+ }
+
+ if (!this.isFailing()) {
+ String msg = "The lane is not failing. This operation should
only be performed if the lane is failing, it should first be allowed to catchup
on its own.";
+ log.warn(msg);
+ return msg;
+ }
+
+ log.info("Running a forced catchup for indexing lane [{}]", name);
+ // First we need to abort and pause the running indexing task
+ this.abortAndPause();
+ log.info("Aborted and paused async indexing for lane [{}]", name);
+ // Release lease for the paused lane
+ this.releaseLeaseForPausedLane();
+ log.info("Released lease for paused lane [{}]", name);
+ String newReferenceCheckpoint = store.checkpoint(lifetime,
ImmutableMap.of(
Review Comment:
Can we use the JDK `Map` instead of the Guava one? `Map.of(...` should be
equivalent.
##########
oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java:
##########
@@ -589,6 +589,83 @@ public void run() {
assertNoConflictMarker(builder);
}
+ @Test
+ public void testForceUpdateAsyncLane() throws CommitFailedException {
+ NodeStore store = new MemoryNodeStore();
+ IndexEditorProvider 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");
+
+ // merge it back in
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ AsyncIndexUpdate async = new AsyncIndexUpdate("async", store,
provider);
+ async.run();
+ NodeState root = store.getRoot();
+ builder = root.builder();
+ builder.child("testRoot1").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ // Run force index catchup with an incorrect confirm message
+ // This will skip the operation and testRoot1 should be indexed in the
next async run.
+ async.getIndexStats().forceIndexLaneCatchup("ok");
+ async.run();
+ root = store.getRoot();
+
+ // first check that the index content nodes exist
+ checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex",
+ INDEX_CONTENT_NODE_NAME);
+ assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode(
+ ":conflict"));
+ PropertyIndexLookup lookup = new PropertyIndexLookup(root);
+ assertEquals(ImmutableSet.of("testRoot", "testRoot1"), find(lookup,
"foo", "abc"));
+
+ // Run force index catchup with correct confirm message
+ // But the async lane is NOT failing
+ // Due to this the force update should be skipped and the
+ // new node testRoot2 will be indexed.
+ builder.child("testRoot2").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ async.getIndexStats().forceIndexLaneCatchup("CONFIRM");
+ async.run();
+ root = store.getRoot();
+
+ checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex",
+ INDEX_CONTENT_NODE_NAME);
+ assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode(
+ ":conflict"));
+ lookup = new PropertyIndexLookup(root);
+ assertEquals(ImmutableSet.of("testRoot", "testRoot1", "testRoot2"),
find(lookup, "foo", "abc"));
+
+
+ // Now run force index update on a failing lane with correct confirm
message
+ builder.child("testRoot3").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ async.getIndexStats().failed(new Exception("Mock index update
failure"));
+ assertTrue(async.isFailing());
+ async.getIndexStats().forceIndexLaneCatchup("CONFIRM");
+ builder.child("testRoot4").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ async.run();
+ root = store.getRoot();
+
+ checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex",
+ INDEX_CONTENT_NODE_NAME);
+ assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode(
+ ":conflict"));
+ lookup = new PropertyIndexLookup(root);
+ // testRoot3 will not be indexed, because it was created after the
last successfully run index update and before the forceUpdate was called.
+ // So it lands in the missing content diff that needs to be reindexed.
+ assertEquals(ImmutableSet.of("testRoot", "testRoot1", "testRoot2",
"testRoot4"), find(lookup, "foo", "abc"));
Review Comment:
Guava vs JDK
##########
oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/AsyncIndexUpdateTest.java:
##########
@@ -589,6 +589,83 @@ public void run() {
assertNoConflictMarker(builder);
}
+ @Test
+ public void testForceUpdateAsyncLane() throws CommitFailedException {
+ NodeStore store = new MemoryNodeStore();
+ IndexEditorProvider 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");
+
+ // merge it back in
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ AsyncIndexUpdate async = new AsyncIndexUpdate("async", store,
provider);
+ async.run();
+ NodeState root = store.getRoot();
+ builder = root.builder();
+ builder.child("testRoot1").setProperty("foo", "abc");
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ // Run force index catchup with an incorrect confirm message
+ // This will skip the operation and testRoot1 should be indexed in the
next async run.
+ async.getIndexStats().forceIndexLaneCatchup("ok");
+ async.run();
+ root = store.getRoot();
+
+ // first check that the index content nodes exist
+ checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex",
+ INDEX_CONTENT_NODE_NAME);
+ assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode(
+ ":conflict"));
+ PropertyIndexLookup lookup = new PropertyIndexLookup(root);
+ assertEquals(ImmutableSet.of("testRoot", "testRoot1"), find(lookup,
"foo", "abc"));
Review Comment:
Again Guava vs JDK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
