This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 3af2c90d63ad CAMEL-25679: camel-mongodb - keep the consumer thread
when a new cursor cannot be opened (#27918)
3af2c90d63ad is described below
commit 3af2c90d63ad4e3a8f80e00beaafd700a6272fc3
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 16:06:09 2026 +0530
CAMEL-25679: camel-mongodb - keep the consumer thread when a new cursor
cannot be opened (#27918)
The tailable and change streams consumers read from a cursor on one thread.
When reading fails (for example the
connection to the server is lost) the thread closes the cursor, waits
cursorRegenerationDelay and opens a new one.
Opening it was done outside the try of the loop: while the server is still
unreachable the driver gives up after its
server selection timeout (30 seconds by default) with a
MongoTimeoutException, which ended the thread. The route stayed
started and consumed nothing until it was restarted; the only trace was the
uncaught exception on stderr.
The new cursor is now opened at the start of the next iteration, inside the
try: a failure is logged like a failed
read ("Will try again on next poll") and tried again after the delay. A
cursor opened while the consumer is being
stopped is closed instead of read, as stop() could not close it. The cursor
field is volatile, as stop() reads it from
another thread.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../mongodb/MongoAbstractConsumerThread.java | 23 +-
.../MongoDbConsumerCursorRegenerationTest.java | 279 +++++++++++++++++++++
2 files changed, 297 insertions(+), 5 deletions(-)
diff --git
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoAbstractConsumerThread.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoAbstractConsumerThread.java
index b1f6cf415a9c..a31ef2cc2f7f 100644
---
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoAbstractConsumerThread.java
+++
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoAbstractConsumerThread.java
@@ -39,7 +39,7 @@ abstract class MongoAbstractConsumerThread implements
Runnable {
protected final MongoCollection<Document> dbCol;
protected final Consumer consumer;
protected final MongoDbEndpoint endpoint;
- protected MongoCursor cursor;
+ protected volatile MongoCursor cursor;
volatile boolean keepRunning = true;
private volatile boolean stopped;
@@ -99,8 +99,20 @@ abstract class MongoAbstractConsumerThread implements
Runnable {
public void run() {
stoppedLatch = new CountDownLatch(1);
try {
- while (keepRunning) {
+ // an interrupt ends the thread, as it did before the cursor was
reopened inside the try
+ while (keepRunning && !Thread.currentThread().isInterrupted()) {
try {
+ if (cursor == null) {
+ // a new cursor after the previous one ended. It is
opened here, inside the try, so that a
+ // failure to open it (the server cannot be reached
yet) is reported and tried again after
+ // the delay, instead of ending this thread while the
consumer stays started
+ cursor = initializeCursor();
+ if (!keepRunning) {
+ // stopped while the cursor was being opened, so
stop() could not close it
+ cursor.close();
+ break;
+ }
+ }
doRun();
} catch (Exception e) {
if (keepRunning) {
@@ -115,7 +127,10 @@ abstract class MongoAbstractConsumerThread implements
Runnable {
}
// regenerate the cursor, if reading failed for some reason
if (keepRunning) {
- cursor.close();
+ if (cursor != null) {
+ cursor.close();
+ cursor = null;
+ }
regeneratingCursor();
if (cursorRegenerationDelayEnabled) {
@@ -126,8 +141,6 @@ abstract class MongoAbstractConsumerThread implements
Runnable {
Thread.currentThread().interrupt();
}
}
-
- cursor = initializeCursor();
}
}
} finally {
diff --git
a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbConsumerCursorRegenerationTest.java
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbConsumerCursorRegenerationTest.java
new file mode 100644
index 000000000000..2909cb1b117c
--- /dev/null
+++
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbConsumerCursorRegenerationTest.java
@@ -0,0 +1,279 @@
+/*
+ * 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.camel.component.mongodb;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.mongodb.CursorType;
+import com.mongodb.MongoSocketReadException;
+import com.mongodb.MongoTimeoutException;
+import com.mongodb.ServerAddress;
+import com.mongodb.client.ChangeStreamIterable;
+import com.mongodb.client.FindIterable;
+import com.mongodb.client.ListCollectionNamesIterable;
+import com.mongodb.client.MongoChangeStreamCursor;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoCursor;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.changestream.ChangeStreamDocument;
+import com.mongodb.client.model.changestream.FullDocument;
+import com.mongodb.client.model.changestream.OperationType;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.bson.BsonDocument;
+import org.bson.BsonInt32;
+import org.bson.Document;
+import org.bson.conversions.Bson;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * When the MongoDB server cannot be reached, reading from the cursor fails
and the consumer thread opens a new cursor
+ * after {@code cursorRegenerationDelay}. Opening it fails too while the
server is still unreachable (the driver gives
+ * up after its server selection timeout, 30 seconds by default). The consumer
thread must then try again, and not end
+ * while the route stays started.
+ */
+class MongoDbConsumerCursorRegenerationTest {
+
+ private static final ServerAddress SERVER = new ServerAddress("localhost",
27017);
+
+ private final MongoClient client = mock(MongoClient.class);
+ private final MongoCollection<Document> collection = mockCollection();
+ private CamelContext context;
+
+ @SuppressWarnings("unchecked")
+ private static MongoCollection<Document> mockCollection() {
+ return mock(MongoCollection.class);
+ }
+
+ @BeforeEach
+ @SuppressWarnings("unchecked")
+ void setUp() {
+ MongoDatabase database = mock(MongoDatabase.class);
+ ListCollectionNamesIterable names =
mock(ListCollectionNamesIterable.class);
+ when(names.spliterator()).thenAnswer(i ->
List.of("events").spliterator());
+ when(database.listCollectionNames()).thenReturn(names);
+ when(database.getCollection("events",
Document.class)).thenReturn(collection);
+ when(database.getName()).thenReturn("test");
+ // the tailable consumer only accepts a capped collection
+ when(database.runCommand(any(Bson.class))).thenReturn(new
Document("capped", true));
+ when(client.getDatabase("test")).thenReturn(database);
+
+ context = new DefaultCamelContext();
+ context.getRegistry().bind("myDb", client);
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (context != null) {
+ context.stop();
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void tailableConsumerOpensANewCursorAfterAFailedRegeneration() throws
Exception {
+ MongoCursor<Document> broken = mock(MongoCursor.class);
+ when(broken.hasNext()).thenThrow(new
MongoSocketReadException("Prematurely reached end of stream", SERVER));
+
+ MongoCursor<Document> recovered = mock(MongoCursor.class);
+ when(recovered.hasNext()).thenReturn(true, false);
+ when(recovered.next()).thenReturn(new Document("increasing",
1).append("text", "after the outage"));
+
+ MongoCursor<Document> idle = mock(MongoCursor.class);
+ when(idle.hasNext()).thenReturn(false);
+
+ FindIterable<Document> find = mock(FindIterable.class);
+ when(collection.find()).thenReturn(find);
+ when(collection.find(any(Bson.class))).thenReturn(find);
+ when(find.cursorType(CursorType.TailableAwait)).thenReturn(find);
+ when(find.iterator())
+ // the cursor of the start, whose read fails when the server
goes away
+ .thenReturn(broken)
+ // the first regeneration, while the server is still
unreachable
+ .thenThrow(new MongoTimeoutException("Timed out while waiting
for a server"))
+ // the server is back
+ .thenReturn(recovered)
+ .thenReturn(idle);
+
+ MockEndpoint mock =
startRoute("mongodb:myDb?database=test&collection=events&tailTrackIncreasingField=increasing"
+ + "&cursorRegenerationDelay=100");
+
+ mock.expectedBodiesReceived(new Document("increasing",
1).append("text", "after the outage"));
+ mock.assertIsSatisfied();
+ verify(find, atLeast(3)).iterator();
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("mongo"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void cursorOpenedWhileTheConsumerStopsIsClosedAndNotRead() throws
Exception {
+ MongoCursor<Document> broken = mock(MongoCursor.class);
+ when(broken.hasNext()).thenThrow(new
MongoSocketReadException("Prematurely reached end of stream", SERVER));
+ MongoCursor<Document> opened = mock(MongoCursor.class);
+
+ // the thread of the tailable consumer, run here on a thread of the
test so that the test can stop it while
+ // it opens a new cursor
+ context.start();
+ MongoDbEndpoint endpoint = context.getEndpoint(
+
"mongodb:myDb?database=test&collection=events&tailTrackIncreasingField=increasing"
+ +
"&cursorRegenerationDelay=100",
+ MongoDbEndpoint.class);
+ MongoDbTailableCursorConsumer consumer
+ = (MongoDbTailableCursorConsumer)
endpoint.createConsumer(exchange -> {
+ });
+ MongoDbTailingThread tailingThread = new
MongoDbTailingThread(endpoint, consumer, consumer.initTailTracking());
+
+ CountDownLatch opening = new CountDownLatch(1);
+ FindIterable<Document> find = mock(FindIterable.class);
+ when(collection.find()).thenReturn(find);
+ when(find.cursorType(CursorType.TailableAwait)).thenReturn(find);
+ when(find.iterator())
+ .thenReturn(broken)
+ .thenAnswer(i -> {
+ // the regeneration is still opening the cursor when the
consumer is stopped
+ opening.countDown();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!tailingThread.keepRunning);
+ return opened;
+ });
+
+ tailingThread.init();
+ Thread thread = new Thread(tailingThread, "tailing");
+ thread.start();
+ assertTrue(opening.await(10, TimeUnit.SECONDS), "the consumer did not
open a new cursor");
+ // returns when the thread has ended
+ tailingThread.stop();
+ thread.join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(thread.isAlive(), "the thread of the consumer has ended");
+
+ // stop() found no cursor to close; the one opened afterwards is
closed by the thread
+ verify(opened).close();
+ verify(opened, never()).hasNext();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void interruptEndsTheThreadWhileTheCursorCannotBeOpened() throws Exception
{
+ context.start();
+ MongoDbEndpoint endpoint = context.getEndpoint(
+
"mongodb:myDb?database=test&collection=events&tailTrackIncreasingField=increasing"
+ +
"&cursorRegenerationDelay=100",
+ MongoDbEndpoint.class);
+ MongoDbTailableCursorConsumer consumer
+ = (MongoDbTailableCursorConsumer)
endpoint.createConsumer(exchange -> {
+ });
+ MongoDbTailingThread tailingThread = new
MongoDbTailingThread(endpoint, consumer, consumer.initTailTracking());
+
+ // the first cursor breaks, then the server cannot be reached: every
attempt to open a new cursor fails
+ MongoCursor<Document> broken = mock(MongoCursor.class);
+ when(broken.hasNext()).thenThrow(new
MongoSocketReadException("Prematurely reached end of stream", SERVER));
+ CountDownLatch attempts = new CountDownLatch(2);
+ FindIterable<Document> find = mock(FindIterable.class);
+ when(collection.find()).thenReturn(find);
+ when(find.cursorType(CursorType.TailableAwait)).thenReturn(find);
+ when(find.iterator())
+ .thenReturn(broken)
+ .thenAnswer(i -> {
+ attempts.countDown();
+ throw new MongoTimeoutException("Timed out while waiting
for a server");
+ });
+
+ tailingThread.init();
+ Thread thread = new Thread(tailingThread, "tailing");
+ thread.start();
+ assertTrue(attempts.await(10, TimeUnit.SECONDS), "the consumer did not
try to open a cursor again");
+ try {
+ thread.interrupt();
+ thread.join(TimeUnit.SECONDS.toMillis(5));
+ assertFalse(thread.isAlive(), "an interrupt ends the thread of the
consumer");
+ assertTrue(tailingThread.keepRunning,
+ "the thread ended because of the interrupt, not because it
was stopped");
+ } finally {
+ // without the guard the thread keeps looping: end it so it does
not outlive the test
+ tailingThread.keepRunning = false;
+ thread.join(TimeUnit.SECONDS.toMillis(5));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void changeStreamsConsumerOpensANewCursorAfterAFailedRegeneration() throws
Exception {
+ MongoChangeStreamCursor<ChangeStreamDocument<Document>> broken =
mock(MongoChangeStreamCursor.class);
+ when(broken.hasNext()).thenThrow(new
MongoSocketReadException("Prematurely reached end of stream", SERVER));
+
+ ChangeStreamDocument<Document> event =
mock(ChangeStreamDocument.class);
+ when(event.getFullDocument()).thenReturn(new Document("_id",
1).append("text", "after the outage"));
+ when(event.getDocumentKey()).thenReturn(new BsonDocument("_id", new
BsonInt32(1)));
+ when(event.getOperationType()).thenReturn(OperationType.INSERT);
+ when(event.getResumeToken()).thenReturn(new BsonDocument("_data", new
BsonInt32(1)));
+ MongoChangeStreamCursor<ChangeStreamDocument<Document>> recovered =
mock(MongoChangeStreamCursor.class);
+ when(recovered.hasNext()).thenReturn(true, false);
+ when(recovered.next()).thenReturn(event);
+
+ MongoChangeStreamCursor<ChangeStreamDocument<Document>> idle =
mock(MongoChangeStreamCursor.class);
+ when(idle.hasNext()).thenReturn(false);
+
+ ChangeStreamIterable<Document> watch =
mock(ChangeStreamIterable.class);
+ when(collection.watch()).thenReturn(watch);
+ when(watch.fullDocument(any(FullDocument.class))).thenReturn(watch);
+ when(watch.resumeAfter(any(BsonDocument.class))).thenReturn(watch);
+ when(watch.iterator())
+ .thenReturn(broken)
+ .thenThrow(new MongoTimeoutException("Timed out while waiting
for a server"))
+ .thenReturn(recovered)
+ .thenReturn(idle);
+
+ MockEndpoint mock =
startRoute("mongodb:myDb?database=test&collection=events&consumerType=changeStreams"
+ + "&cursorRegenerationDelay=100");
+
+ mock.expectedBodiesReceived(new Document("_id", 1).append("text",
"after the outage"));
+ mock.assertIsSatisfied();
+ verify(watch, atLeast(3)).iterator();
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("mongo"));
+ }
+
+ private MockEndpoint startRoute(String uri) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri).routeId("mongo").to("mock:result");
+ }
+ });
+ context.start();
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
+ mock.setResultWaitTime(10000);
+ return mock;
+ }
+}
