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 th...@apache.org on 2026/10/20 12:03:39 UTC
(jackrabbit-oak) branch trunk updated: OAK-12466 Oak-run indexing can miss entries with mongoParallelDump enabled (#3221)
This is an automated email from the ASF dual-hosted git repository.
thomasmueller 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 15e7b6ad48 OAK-12466 Oak-run indexing can miss entries with mongoParallelDump enabled (#3221)
15e7b6ad48 is described below
commit 15e7b6ad4863f7e6d667f77bd361349a8ba4295e
Author: Thomas Mueller <th...@apache.org>
AuthorDate: Tue Oct 20 14:03:33 2026 +0200
OAK-12466 Oak-run indexing can miss entries with mongoParallelDump enabled (#3221)
* OAK-12466 Oak-run indexing can miss entries with mongoParallelDump enabled
* OAK-12466 Oak-run indexing can miss entries with mongoParallelDump enabled
---
.../src/site/markdown/query/oak-run-indexing.md | 6 +
.../pipelined/PipelinedMongoDownloadTask.java | 19 +-
...PipelinedMongoDownloadTaskCancellationTest.java | 260 +++++++++++++++++++++
3 files changed, 284 insertions(+), 1 deletion(-)
diff --git a/oak-doc/src/site/markdown/query/oak-run-indexing.md b/oak-doc/src/site/markdown/query/oak-run-indexing.md
index 551ba76685..b289302d8c 100644
--- a/oak-doc/src/site/markdown/query/oak-run-indexing.md
+++ b/oak-doc/src/site/markdown/query/oak-run-indexing.md
@@ -116,6 +116,12 @@ Supported for only Lucene indexes.
If the indexes being reindex have fulltext indexing enabled then refer to [Tika Setup](#tika-setup) for steps
on how to adapt the command to include Tika support for text extraction
+When parallel MongoDB downloading is enabled with `oak.indexer.pipelined.mongoParallelDump=true`,
+both download streams must finish delivering their buffered documents before indexing proceeds.
+After cancelling a stream, the indexer waits up to `oak.indexer.pipelined.mongoConnectionRetrySeconds`
+(300 seconds by default) for it to finish. If it times out or fails, indexing fails rather than
+continuing with potentially missing entries.
+
### <a name="out-of-band-indexing"></a> A - out-of-band indexing
Out of band indexing has following phases
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java
index 626c553cf6..443dd55a5e 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java
@@ -125,6 +125,9 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
*/
public static final String OAK_INDEXER_PIPELINED_RETRY_ON_CONNECTION_ERRORS = "oak.indexer.pipelined.retryOnConnectionErrors";
public static final boolean DEFAULT_OAK_INDEXER_PIPELINED_RETRY_ON_CONNECTION_ERRORS = true;
+ /**
+ * Connection retry period and maximum wait for a cancelled parallel download to flush its pending documents.
+ */
public static final String OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = "oak.indexer.pipelined.mongoConnectionRetrySeconds";
public static final int DEFAULT_OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = 300;
@@ -443,7 +446,8 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
ancestorsDownloadTask.downloadAncestors(mongoFilterPaths.included);
}
if (parallelDump) {
- LOG.info("Downloading in parallel with two connections, one in ascending the other in descending order");
+ LOG.info("Downloading in parallel with two connections, one in ascending the other in descending order. " +
+ "Wait for cancelled downloads to flush.");
// Shared object between the two download threads to store the last document download by either of them
// and determine when the two downloads have crossed
MongoParallelDownloadCoordinator parallelDownloadCoordinator = new MongoParallelDownloadCoordinator();
@@ -497,15 +501,18 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
// If the task failed with an exception, the exception will be thrown by get()
completedTask.get();
// One of the download tasks has completed. Cancel the other one.
+ Future<?> cancelledDownloadFuture;
if (completedTask == ascendingDownloadFuture) {
LOG.info("Ascending download task has completed. Cancelling descending download task.");
// This will set a flag indicating that the download should be cancelled and close the Mongo
// cursor, which will cause the download task to abort next time it performs an operation on
// the cursor or if it is blocked on the cursor.
descendingDownloadTask.cancelDownload();
+ cancelledDownloadFuture = descendingDownloadFuture;
} else if (completedTask == descendingDownloadFuture) {
LOG.info("Descending download task has completed. Cancelling ascending download task.");
ascendingDownloadTask.cancelDownload();
+ cancelledDownloadFuture = ascendingDownloadFuture;
} else {
throw new IllegalStateException("Unknown download task completed: " + completedTask);
}
@@ -531,6 +538,16 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
LOG.info("Waiting for download of updated documents to complete.");
updateDocsDownloadFuture.get();
}
+ // Crossing includes progress from documents still buffered by the cancelled task.
+ LOG.info("Waiting for cancelled download task to finish flushing pending documents.");
+ try {
+ cancelledDownloadFuture.get(retryDuringSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ TimeoutException timeout = new TimeoutException("Cancelled MongoDB download did not finish " +
+ "flushing pending documents within " + retryDuringSeconds + " seconds");
+ timeout.initCause(e);
+ throw timeout;
+ }
return;
}
}
diff --git a/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTaskCancellationTest.java b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTaskCancellationTest.java
new file mode 100644
index 0000000000..2b65707257
--- /dev/null
+++ b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTaskCancellationTest.java
@@ -0,0 +1,260 @@
+/*
+ * 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.index.indexer.document.flatfile.pipelined;
+
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.client.FindIterable;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoCursor;
+import com.mongodb.client.MongoDatabase;
+import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore;
+import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStoreHelper;
+import org.apache.jackrabbit.oak.plugins.index.IndexingReporter;
+import org.apache.jackrabbit.oak.spi.filter.PathFilter;
+import org.apache.jackrabbit.oak.stats.StatisticsProvider;
+import org.bson.RawBsonDocument;
+import org.bson.conversions.Bson;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.mockito.Answers;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedMongoDownloadTask.SENTINEL_MONGO_DOCUMENT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+public class PipelinedMongoDownloadTaskCancellationTest {
+ @Rule
+ public final RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
+
+ @Rule
+ public final MockitoRule mocks = MockitoJUnit.rule();
+
+ @Mock
+ private MongoClient client;
+ @Mock
+ private MongoDatabase database;
+ @Mock
+ private MongoCollection<RawBsonDocument> collection;
+ @Mock(answer = Answers.RETURNS_SELF)
+ private FindIterable<RawBsonDocument> iterable;
+ @Mock
+ private MongoCursor<RawBsonDocument> ascending;
+ @Mock
+ private MongoCursor<RawBsonDocument> descending;
+ @Mock
+ private MongoCursor<RawBsonDocument> updated;
+ @Mock
+ private MongoCursor<RawBsonDocument> empty;
+
+ @Test(timeout = 20000)
+ public void flushCancelledDescendingDownloadBeforeSentinel() throws Exception {
+ cancelledDownload(false, CancellationResult.FLUSH);
+ }
+
+ @Test(timeout = 20000)
+ public void flushCancelledAscendingDownloadBeforeSentinel() throws Exception {
+ cancelledDownload(true, CancellationResult.FLUSH);
+ }
+
+ @Test(timeout = 20000)
+ public void cancelledDownloadFailureAbortsPipeline() throws Exception {
+ cancelledDownload(false, CancellationResult.FAILURE);
+ }
+
+ @Test(timeout = 20000)
+ public void cancelledDownloadTimeoutAbortsPipeline() throws Exception {
+ cancelledDownload(false, CancellationResult.TIMEOUT);
+ }
+
+ private enum CancellationResult {
+ FLUSH, FAILURE, TIMEOUT
+ }
+
+ private void cancelledDownload(boolean cancelAscending, CancellationResult cancellationResult) throws Exception {
+ System.setProperty(PipelinedMongoDownloadTask.OAK_INDEXER_PIPELINED_MONGO_PARALLEL_DUMP, "true");
+ System.setProperty(PipelinedMongoDownloadTask.OAK_INDEXER_PIPELINED_MONGO_PARALLEL_DUMP_SECONDARIES_ONLY, "false");
+ System.setProperty(PipelinedMongoDownloadTask.OAK_INDEXER_PIPELINED_MONGO_REGEX_PATH_FILTERING, "true");
+ System.setProperty(PipelinedMongoDownloadTask.OAK_INDEXER_PIPELINED_RETRY_ON_CONNECTION_ERRORS, "true");
+ System.setProperty(PipelinedMongoDownloadTask.OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS,
+ cancellationResult == CancellationResult.TIMEOUT ? "1" : "5");
+
+ when(client.getDatabase("oak")).thenReturn(database);
+ when(database.getCollection("nodes", RawBsonDocument.class)).thenReturn(collection);
+ when(collection.find(any(Bson.class))).thenReturn(iterable);
+ when(iterable.iterator()).thenAnswer(invocation -> {
+ switch (Thread.currentThread().getName()) {
+ case "mongo-dump-ascending":
+ return ascending;
+ case "mongo-dump-descending":
+ return descending;
+ case "mongo-dump-updated-docs":
+ return updated;
+ default:
+ return empty;
+ }
+ });
+
+ CountDownLatch buffered = new CountDownLatch(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ CompletableFuture<Void> releaseCursor = new CompletableFuture<>();
+ CountDownLatch sentinel = new CountDownLatch(1);
+ LinkedBlockingQueue<RawBsonDocument[]> queue = new LinkedBlockingQueue<>() {
+ @Override
+ public void put(RawBsonDocument[] documents) throws InterruptedException {
+ if (documents == SENTINEL_MONGO_DOCUMENT) {
+ sentinel.countDown();
+ }
+ super.put(documents);
+ }
+ };
+ MongoCursor<RawBsonDocument> blockedCursor = cancelAscending ? ascending : descending;
+ MongoCursor<RawBsonDocument> crossingCursor = cancelAscending ? descending : ascending;
+ long modified = cancelAscending ? 10 : 20;
+ RawBsonDocument bufferedDocument = document("buffered", modified);
+ RawBsonDocument crossingDocument = document("crossing", modified + (cancelAscending ? -1 : 1));
+ RawBsonDocument newestDocument = document("newest", 30);
+ IllegalArgumentException cursorFailure = new IllegalArgumentException("Cancelled cursor failed");
+ Iterator<RawBsonDocument> pendingDocuments = (cancelAscending ?
+ List.of(bufferedDocument) : List.of(newestDocument, bufferedDocument)).iterator();
+ when(blockedCursor.next()).thenAnswer(invocation -> pendingDocuments.next());
+ when(blockedCursor.hasNext()).thenAnswer(invocation -> {
+ if (pendingDocuments.hasNext()) {
+ return true;
+ }
+ buffered.countDown();
+ // An in-flight getMore may outlive both cursor.close() and a thread interruption.
+ releaseCursor.join();
+ if (cancellationResult == CancellationResult.FAILURE) {
+ throw cursorFailure;
+ }
+ return false;
+ });
+ doAnswer(invocation -> {
+ cancelled.countDown();
+ return null;
+ }).when(blockedCursor).close();
+ Iterator<RawBsonDocument> crossingDocuments = (cancelAscending ?
+ List.of(newestDocument, crossingDocument) : List.of(crossingDocument)).iterator();
+ when(crossingCursor.hasNext()).thenAnswer(invocation -> {
+ assertTrue("Other stream did not buffer its document", buffered.await(10, TimeUnit.SECONDS));
+ return crossingDocuments.hasNext();
+ });
+ when(crossingCursor.next()).thenAnswer(invocation -> crossingDocuments.next());
+ when(updated.hasNext()).thenReturn(true, true, false);
+ when(updated.next()).thenReturn(newestDocument);
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ Future<?> release = executor.submit(() -> {
+ try {
+ assertTrue("Buffered stream was not cancelled", cancelled.await(10, TimeUnit.SECONDS));
+ if (cancellationResult == CancellationResult.FLUSH) {
+ assertFalse("End-of-input overtook the cancelled stream's buffered document",
+ sentinel.await(2, TimeUnit.SECONDS));
+ }
+ } finally {
+ if (cancellationResult != CancellationResult.TIMEOUT) {
+ releaseCursor.complete(null);
+ }
+ }
+ return null;
+ });
+ String originalThreadName = Thread.currentThread().getName();
+ List<Thread> downloadThreads = new CopyOnWriteArrayList<>();
+ MongoDocumentStore store = mock(MongoDocumentStore.class);
+ try (MockedStatic<MongoClients> clients = mockStatic(MongoClients.class);
+ MockedStatic<MongoDocumentStoreHelper> helper = mockStatic(MongoDocumentStoreHelper.class)) {
+ clients.when(() -> MongoClients.create(any(MongoClientSettings.class))).thenReturn(client);
+ helper.when(() -> MongoDocumentStoreHelper.getMongoDatabaseName(store)).thenReturn("oak");
+ PipelinedMongoDownloadTask task = new PipelinedMongoDownloadTask(
+ new ConnectionString("mongodb://localhost"), store, 4 * 1024 * 1024, 10000, queue,
+ List.of(new PathFilter(List.of("/a/b"), List.of())), StatisticsProvider.NOOP,
+ mock(IndexingReporter.class), runnable -> {
+ Thread thread = new Thread(runnable);
+ downloadThreads.add(thread);
+ return thread;
+ });
+
+ if (cancellationResult == CancellationResult.FLUSH) {
+ PipelinedMongoDownloadTask.Result result = task.call();
+ List<RawBsonDocument> downloaded = new ArrayList<>();
+ RawBsonDocument[] batch;
+ while ((batch = queue.poll()) != SENTINEL_MONGO_DOCUMENT) {
+ assertNotNull("Download did not signal end-of-input", batch);
+ downloaded.addAll(List.of(batch));
+ }
+ assertEquals(cancelAscending ?
+ List.of(newestDocument, crossingDocument, newestDocument, bufferedDocument) :
+ List.of(crossingDocument, newestDocument, newestDocument, bufferedDocument), downloaded);
+ assertEquals(downloaded.size(), result.getDocumentsDownloaded());
+ assertTrue("Documents were delivered after end-of-input", queue.isEmpty());
+ } else {
+ if (cancellationResult == CancellationResult.FAILURE) {
+ RuntimeException failure = assertThrows(RuntimeException.class, task::call);
+ assertSame(cursorFailure, failure.getCause().getCause());
+ } else {
+ TimeoutException failure = assertThrows(TimeoutException.class, task::call);
+ assertTrue(failure.getMessage().contains("flushing pending documents within 1 seconds"));
+ }
+ assertEquals("Failed download must not signal end-of-input", 1, sentinel.getCount());
+ }
+ release.get(10, TimeUnit.SECONDS);
+ } finally {
+ Thread.currentThread().setName(originalThreadName);
+ releaseCursor.complete(null);
+ executor.shutdownNow();
+ assertTrue("Cursor controller did not terminate", executor.awaitTermination(10, TimeUnit.SECONDS));
+ for (Thread thread : downloadThreads) {
+ thread.join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse("Download worker did not terminate", thread.isAlive());
+ }
+ }
+ }
+
+ private static RawBsonDocument document(String name, long modified) {
+ return RawBsonDocument.parse("{_id: '3:/a/b/" + name + "', _modified: {$numberLong: '" + modified + "'}}");
+ }
+}