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 33accc67f274 CAMEL-25731: camel-mongodb - log the stack trace only for
the first of consecutive cursor failures (#29294)
33accc67f274 is described below
commit 33accc67f274b3af49e2cd011f1a24836c9049b5
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 6 13:06:39 2026 +0530
CAMEL-25731: camel-mongodb - log the stack trace only for the first of
consecutive cursor failures (#29294)
The tailing consumer logs the stack trace only for the first of consecutive
cursor failures, later ones as a one-line WARN (stack trace at DEBUG), and logs
the failure count at INFO once a cursor opens again.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../mongodb/MongoAbstractConsumerThread.java | 23 ++++++--
.../MongoDbConsumerCursorRegenerationTest.java | 65 ++++++++++++++++++++++
2 files changed, 84 insertions(+), 4 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 a31ef2cc2f7f..94d4e9bdab30 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
@@ -98,6 +98,9 @@ abstract class MongoAbstractConsumerThread implements
Runnable {
@Override
public void run() {
stoppedLatch = new CountDownLatch(1);
+ // the failures since a cursor was last opened: while the server
cannot be reached, every attempt fails the same
+ // way, so only the first one logs the stack trace
+ int failures = 0;
try {
// an interrupt ends the thread, as it did before the cursor was
reopened inside the try
while (keepRunning && !Thread.currentThread().isInterrupted()) {
@@ -112,14 +115,26 @@ abstract class MongoAbstractConsumerThread implements
Runnable {
cursor.close();
break;
}
+ if (failures > 0) {
+ log.info("Opened a new cursor to consume from
MongoDB again (consecutive failures: {})", failures);
+ failures = 0;
+ }
}
doRun();
} catch (Exception e) {
if (keepRunning) {
- // this one repeats for as long as the cause persists,
so it is the branch that
- // needs the stack trace, not the one below
- log.warn("Exception from consuming from MongoDB caused
by {}. Will try again on next poll.",
- e.getMessage(), e);
+ failures++;
+ if (failures == 1) {
+ log.warn("Exception from consuming from MongoDB
caused by {}. Will try again on next poll.",
+ e.getMessage(), e);
+ } else {
+ // this one repeats for as long as the cause
persists: the stack trace was logged with the
+ // first failure
+ log.warn("Exception from consuming from MongoDB
caused by {} ({} consecutive failures)."
+ + " Will try again on next poll.",
+ e.getMessage(), failures);
+ log.debug("Exception from consuming from MongoDB",
e);
+ }
} else {
log.warn("Exception from consuming from MongoDB caused
by {}. ConsumerThread will be stopped.",
e.getMessage(), e);
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
index 2909cb1b117c..b0fe853f305b 100644
---
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
@@ -17,6 +17,7 @@
package org.apache.camel.component.mongodb;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -40,6 +41,12 @@ 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.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
import org.bson.BsonDocument;
import org.bson.BsonInt32;
import org.bson.Document;
@@ -227,6 +234,64 @@ class MongoDbConsumerCursorRegenerationTest {
}
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void stackTraceIsLoggedOnlyForTheFirstOfConsecutiveFailures() throws
Exception {
+ context.start();
+ MongoDbEndpoint endpoint = context.getEndpoint(
+
"mongodb:myDb?database=test&collection=events&tailTrackIncreasingField=increasing"
+ +
"&cursorRegenerationDelay=10",
+ MongoDbEndpoint.class);
+ MongoDbTailableCursorConsumer consumer
+ = (MongoDbTailableCursorConsumer)
endpoint.createConsumer(exchange -> {
+ });
+ MongoDbTailingThread tailingThread = new
MongoDbTailingThread(endpoint, consumer, consumer.initTailTracking());
+
+ // the cursor breaks, opening a new one fails three times while the
server cannot be reached, then it is back
+ MongoCursor<Document> broken = mock(MongoCursor.class);
+ when(broken.hasNext()).thenThrow(new
MongoSocketReadException("Prematurely reached end of stream", SERVER));
+ MongoCursor<Document> idle = mock(MongoCursor.class);
+ when(idle.hasNext()).thenReturn(false);
+ MongoTimeoutException unreachable = new MongoTimeoutException("Timed
out while waiting for a server");
+ FindIterable<Document> find = mock(FindIterable.class);
+ when(collection.find()).thenReturn(find);
+ when(find.cursorType(CursorType.TailableAwait)).thenReturn(find);
+ when(find.iterator()).thenReturn(broken).thenThrow(unreachable,
unreachable, unreachable).thenReturn(idle);
+
+ List<LogEvent> events = new CopyOnWriteArrayList<>();
+ AbstractAppender appender = new AbstractAppender("cursorFailures",
null, null, true, Property.EMPTY_ARRAY) {
+ @Override
+ public void append(LogEvent event) {
+ events.add(event.toImmutable());
+ }
+ };
+ appender.start();
+ Logger logger = (Logger)
LogManager.getLogger(MongoDbTailingThread.class);
+ logger.addAppender(appender);
+ tailingThread.init();
+ Thread thread = new Thread(tailingThread, "tailing");
+ try {
+ thread.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
events.stream().anyMatch(e -> e.getLevel() == Level.INFO
+ && e.getMessage().getFormattedMessage().startsWith("Opened
a new cursor")));
+ } finally {
+ tailingThread.stop();
+ thread.join(TimeUnit.SECONDS.toMillis(10));
+ logger.removeAppender(appender);
+ appender.stop();
+ }
+
+ assertFalse(thread.isAlive(), "the thread ends when stopped");
+ List<LogEvent> warnings = events.stream().filter(e -> e.getLevel() ==
Level.WARN).toList();
+ assertEquals(4, warnings.size(), "one WARN per failure");
+ assertEquals(1, warnings.stream().filter(e -> e.getThrown() !=
null).count(),
+ "only the first of the consecutive failures logs the stack
trace");
+ assertTrue(warnings.get(0).getThrown() instanceof
MongoSocketReadException);
+ List<String> opened = events.stream().filter(e -> e.getLevel() ==
Level.INFO)
+ .map(e -> e.getMessage().getFormattedMessage()).filter(m ->
m.startsWith("Opened a new cursor")).toList();
+ assertEquals(List.of("Opened a new cursor to consume from MongoDB
again (consecutive failures: 4)"), opened);
+ }
+
@Test
@SuppressWarnings("unchecked")
void changeStreamsConsumerOpensANewCursorAfterAFailedRegeneration() throws
Exception {
