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 1a4077f0c46f CAMEL-25711: camel-mongodb-gridfs - keep polling after a
poll fails (#27991)
1a4077f0c46f is described below
commit 1a4077f0c46f2367aaca8a26aedc8dcb0480accd
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 14:42:51 2026 +0530
CAMEL-25711: camel-mongodb-gridfs - keep polling after a poll fails (#27991)
GridFsConsumer polls with a BackgroundTask without a duration limit. Only
the processing of the exchange was inside a try: an exception of the
bucket query, the cursor, the file attribute claim, openDownloadStream
or the persistent timestamp update (for example a MongoTimeoutException
while MongoDB cannot be reached) left the task, so BackgroundTask marked
it failed and the poll was never scheduled again. The consumer stayed
started, routed nothing and logged nothing.
The task now calls poll(), which reports the failure through the
consumer's ExceptionHandler (the first of a run of failures; the next
ones at DEBUG, and an INFO line when polling works again) and returns
false, so the next poll tries again after the delay. The persistent
timestamp is read by the first poll that reaches MongoDB instead of by
run(), so a failure there is retried as well instead of leaving the
consumer without any poll.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/mongodb/gridfs/GridFsConsumer.java | 96 ++++++++++----
.../gridfs/GridFsConsumerTimestampTest.java | 138 ++++++++++++++++++++-
2 files changed, 205 insertions(+), 29 deletions(-)
diff --git
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
index 40afc3d4f142..2e6717382c23 100644
---
a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
+++
b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
@@ -39,6 +39,8 @@ import org.apache.camel.support.task.Tasks;
import org.apache.camel.support.task.budget.Budgets;
import org.bson.Document;
import org.bson.conversions.Bson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static com.mongodb.client.model.Filters.eq;
import static
org.apache.camel.component.mongodb.gridfs.GridFsConstants.GRIDFS_FILE_ATTRIBUTE_DONE;
@@ -48,6 +50,8 @@ import static
org.apache.camel.component.mongodb.gridfs.GridFsConstants.GRIDFS_F
import static
org.apache.camel.component.mongodb.gridfs.GridFsConstants.PERSISTENT_TIMESTAMP_KEY;
public class GridFsConsumer extends DefaultConsumer implements Runnable {
+ private static final Logger LOG =
LoggerFactory.getLogger(GridFsConsumer.class);
+
private final GridFsEndpoint endpoint;
private volatile ScheduledExecutorService taskExecutor;
private volatile ExecutorService executor;
@@ -58,6 +62,12 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
// Two threads use it: run() sets the start value on the thread of the
executor, before it schedules the polls;
// the polls then read and move it on the thread of the taskExecutor.
private volatile Date fromDate;
+ // the collection and the document of the persistent timestamp, read by
the first poll that can reach MongoDB.
+ // run() resets them before it schedules the polls, the polls then use
them on the thread of the taskExecutor
+ private volatile MongoCollection<Document> ptsCollection;
+ private volatile Document persistentTimestamp;
+ // the number of polls in a row that failed, so that only the first
failure of an outage is reported
+ private volatile int consecutiveFailures;
public GridFsConsumer(GridFsEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -90,6 +100,9 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
@Override
public void run() {
fromDate = null;
+ // the start of the consumer: the first value of the timestamp, also
of a persistent timestamp that the first
+ // poll that reaches MongoDB creates
+ Date startDate = new Date();
QueryStrategy queryStrategy = endpoint.getQueryStrategy();
boolean usesTimestamp = queryStrategy != QueryStrategy.FileAttribute;
@@ -99,26 +112,11 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
|| queryStrategy == QueryStrategy.TimeStampAndFileAttribute
|| queryStrategy ==
QueryStrategy.PersistentTimestampAndFileAttribute;
- MongoCollection<Document> ptsCollection = null;
- Document persistentTimestamp = null;
- if (persistsTimestamp) {
- ptsCollection =
endpoint.getDB().getCollection(endpoint.getPersistentTSCollection());
-
- // ensure standard indexes as long as collections are small
- if (ptsCollection.countDocuments() < 1000) {
- ptsCollection.createIndex(new BasicDBObject("id", 1));
- }
-
- persistentTimestamp = ptsCollection.find(eq("id",
endpoint.getPersistentTSObject())).first();
- if (persistentTimestamp == null) {
- persistentTimestamp = new Document("id",
endpoint.getPersistentTSObject());
- fromDate = new Date();
- persistentTimestamp.put(PERSISTENT_TIMESTAMP_KEY, fromDate);
- ptsCollection.insertOne(persistentTimestamp);
- }
- fromDate = persistentTimestamp.get(PERSISTENT_TIMESTAMP_KEY,
Date.class);
- } else if (usesTimestamp) {
- fromDate = new Date();
+ ptsCollection = null;
+ persistentTimestamp = null;
+ consecutiveFailures = 0;
+ if (usesTimestamp && !persistsTimestamp) {
+ fromDate = startDate;
}
BlockingTask task = Tasks.backgroundTask()
@@ -132,11 +130,59 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
.withName("GridFsPolling")
.build();
- MongoCollection<Document> finalPtsCollection = ptsCollection;
- Document finalPersistentTimestamp = persistentTimestamp;
- task.run(endpoint.getCamelContext(),
- () -> processCollection(usesTimestamp, persistsTimestamp,
usesAttribute, finalPtsCollection,
- finalPersistentTimestamp));
+ task.run(endpoint.getCamelContext(), () -> poll(usesTimestamp,
persistsTimestamp, usesAttribute, startDate));
+ }
+
+ /**
+ * One poll. A failure, such as MongoDB that cannot be reached, is
reported and the next poll tries again: an
+ * exception thrown out of the repeated task would end it, and the
consumer would stay started without polling.
+ */
+ private boolean poll(boolean usesTimestamp, boolean persistsTimestamp,
boolean usesAttribute, Date startDate) {
+ try {
+ if (persistsTimestamp && persistentTimestamp == null) {
+ readPersistentTimestamp(startDate);
+ }
+ boolean answer = processCollection(usesTimestamp,
persistsTimestamp, usesAttribute, ptsCollection,
+ persistentTimestamp);
+ if (consecutiveFailures > 0) {
+ LOG.info("Polling {} works again after {} failed polls",
endpoint, consecutiveFailures);
+ consecutiveFailures = 0;
+ }
+ return answer;
+ } catch (Exception e) {
+ if (!isRunAllowed()) {
+ // the consumer is stopping (for example the MongoClient is
closed during shutdown): no poll follows
+ LOG.debug("Error polling {} while the consumer is stopping:
{}", endpoint, e.toString());
+ return false;
+ }
+ consecutiveFailures++;
+ if (consecutiveFailures == 1) {
+ getExceptionHandler().handleException("Error polling " +
endpoint + " (will try again on the next poll)", e);
+ } else {
+ // the failure of the outage is reported already, and the
polls run every delay (500 ms by default)
+ LOG.debug("Error polling {} ({} polls in a row failed): {}",
endpoint, consecutiveFailures, e.getMessage());
+ }
+ return false;
+ }
+ }
+
+ private void readPersistentTimestamp(Date startDate) {
+ MongoCollection<Document> collection =
endpoint.getDB().getCollection(endpoint.getPersistentTSCollection());
+
+ // ensure standard indexes as long as collections are small
+ if (collection.countDocuments() < 1000) {
+ collection.createIndex(new BasicDBObject("id", 1));
+ }
+
+ Document timestamp = collection.find(eq("id",
endpoint.getPersistentTSObject())).first();
+ if (timestamp == null) {
+ timestamp = new Document("id", endpoint.getPersistentTSObject());
+ timestamp.put(PERSISTENT_TIMESTAMP_KEY, startDate);
+ collection.insertOne(timestamp);
+ }
+ fromDate = timestamp.get(PERSISTENT_TIMESTAMP_KEY, Date.class);
+ ptsCollection = collection;
+ persistentTimestamp = timestamp;
}
private boolean processCollection(
diff --git
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
index 47a214d76dfc..4994049fb485 100644
---
a/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
+++
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
@@ -25,8 +25,11 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoTimeoutException;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
@@ -37,8 +40,10 @@ import com.mongodb.client.gridfs.GridFSFindIterable;
import com.mongodb.client.gridfs.model.GridFSFile;
import com.mongodb.client.model.FindOneAndUpdateOptions;
import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.ExceptionHandler;
import org.bson.BsonArray;
import org.bson.BsonDocument;
import org.bson.BsonObjectId;
@@ -52,6 +57,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
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.ArgumentMatchers.anyString;
@@ -72,6 +78,20 @@ class GridFsConsumerTimestampTest {
private final List<Date> persisted = new CopyOnWriteArrayList<>();
// the ids of the files that the consumer has marked with its file
attribute
private final Set<BsonValue> marked = ConcurrentHashMap.newKeySet();
+ // the number of polls (queries of the bucket) that fail, as when MongoDB
cannot be reached
+ private final AtomicInteger failingPolls = new AtomicInteger();
+ // the number of reads of the persistent timestamp that fail
+ private final AtomicInteger failingTimestampReads = new AtomicInteger();
+ // called when a read of the persistent timestamp fails
+ private volatile Runnable onFailedTimestampRead = () -> {
+ };
+ // the persistent timestamp documents that the consumer created
+ private final List<Date> created = new CopyOnWriteArrayList<>();
+ // the failures that the consumer reported through its ExceptionHandler
+ private final List<Throwable> reported = new CopyOnWriteArrayList<>();
+ // called by a query of the bucket, before the files are returned (or the
failure is thrown)
+ private volatile Runnable onPoll = () -> {
+ };
private final GridFSBucket bucket = mock(GridFSBucket.class);
private final MongoDatabase database = mock(MongoDatabase.class);
@SuppressWarnings("unchecked")
@@ -87,6 +107,11 @@ class GridFsConsumerTimestampTest {
context.start();
when(bucket.find(any(Bson.class))).thenAnswer(invocation -> {
+ onPoll.run();
+ if (failingPolls.getAndUpdate(n -> Math.max(0, n - 1)) > 0) {
+ polls.countDown();
+ throw new MongoTimeoutException("Timed out while waiting for a
server (test)");
+ }
List<GridFSFile> found =
filesUploadedAfter(invocation.getArgument(0));
polls.countDown();
return iterableOf(found);
@@ -180,7 +205,82 @@ class GridFsConsumerTimestampTest {
"the file attribute makes sure a file is routed once, the
query keeps the start date");
}
+ @ParameterizedTest
+ @EnumSource(QueryStrategy.class)
+ void consumerKeepsPollingAfterAFailedPoll(QueryStrategy strategy) throws
Exception {
+ failingPolls.set(2);
+ startConsumer(strategy);
+
+ // the two failing polls, then one that sees no file
+ awaitFullPolls(2);
+ stored.add(file("first", 60));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("first"), routed, "a failed poll (MongoDB not
reachable) must not end the consumer");
+ }
+
+ @Test
+ void
persistentTimestampStrategyKeepsPollingWhenTheTimestampCannotBeReadAtStart()
throws Exception {
+ failingTimestampReads.set(1);
+ startConsumer(QueryStrategy.PersistentTimestamp);
+
+ stored.add(file("first", 60));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("first"), routed,
+ "the timestamp could not be read when the consumer started,
the next poll must read it");
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = QueryStrategy.class, names = { "PersistentTimestamp",
"PersistentTimestampAndFileAttribute" })
+ void
newPersistentTimestampStartsWhenTheConsumerStartsAlthoughTheFirstPollFails(QueryStrategy
strategy)
+ throws Exception {
+ // the first poll cannot read the timestamp (MongoDB not reachable
yet); meanwhile another client uploads a file
+ AtomicReference<Date> failedAt = new AtomicReference<>();
+ failingTimestampReads.set(1);
+ onFailedTimestampRead = () -> {
+ failedAt.set(new Date());
+ stored.add(new GridFSFile(new BsonObjectId(), "early", 1, 1, new
Date(failedAt.get().getTime() + 25), null));
+ };
+ // the next poll, which creates the timestamp, runs at least 50 ms
after the failed one
+ startConsumer(strategy, 50);
+
+ awaitFullPolls(2);
+
+ assertEquals(1, created.size());
+ assertFalse(created.get(0).after(failedAt.get()),
+ "the new timestamp must be the start of the consumer, not the
time of the first poll that worked");
+ assertEquals(List.of("early"), routed, "a file uploaded after the
consumer started must be routed");
+ }
+
+ @Test
+ void pollFailureWhileTheConsumerStopsIsNotReported() throws Exception {
+ AtomicReference<Thread> pollThread = new AtomicReference<>();
+ CountDownLatch stopped = new CountDownLatch(1);
+ startConsumer(QueryStrategy.TimeStamp);
+ awaitFullPolls(1);
+
+ // the consumer is stopped during a poll, which then fails (as when
the MongoClient is closed by the shutdown)
+ onPoll = () -> {
+ if (pollThread.compareAndSet(null, Thread.currentThread())) {
+ consumer.stop();
+ stopped.countDown();
+ throw new IllegalStateException("state should be: open
(test)");
+ }
+ };
+ assertTrue(stopped.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "the
consumer did not poll again");
+ // the polling thread ends once the failed poll returns, as its
executor is shut down
+ pollThread.get().join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertFalse(pollThread.get().isAlive(), "the failed poll did not
return");
+
+ assertEquals(List.of(), reported, "no poll follows, the failure must
not be reported as one that is retried");
+ }
+
private void startConsumer(QueryStrategy strategy) {
+ startConsumer(strategy, 1);
+ }
+
+ private void startConsumer(QueryStrategy strategy, long delay) {
GridFsComponent component = new GridFsComponent();
component.setCamelContext(context);
GridFsEndpoint endpoint = new GridFsEndpoint("mongodb-gridfs:fake",
component) {
@@ -202,11 +302,27 @@ class GridFsConsumerTimestampTest {
endpoint.setCamelContext(context);
endpoint.setQueryStrategy(strategy);
endpoint.setInitialDelay(1);
- endpoint.setDelay(1);
+ endpoint.setDelay(delay);
Processor processor = exchange ->
routed.add(exchange.getMessage().getBody(GridFSDownloadStream.class)
.getGridFSFile().getFilename());
consumer = new GridFsConsumer(endpoint, processor);
+ consumer.setExceptionHandler(new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ reported.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ reported.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ reported.add(exception);
+ }
+ });
consumer.start();
}
@@ -229,12 +345,14 @@ class GridFsConsumerTimestampTest {
}
/**
- * Evaluates the query of the consumer on the stored files:
<tt>{uploadDate: {$gt: date}}</tt>, and with a file
- * attribute also <tt>{camel-processed: null}</tt>.
+ * Evaluates the query of the consumer on the stored files:
<tt>{uploadDate: {$gt: date}}</tt> (not with the
+ * FileAttribute strategy), and with a file attribute also
<tt>{camel-processed: null}</tt>.
*/
private List<GridFSFile> filesUploadedAfter(Bson query) {
BsonDocument document = toDocument(query);
- long after = find(document,
GridFsConstants.GRIDFS_FILE_KEY_UPLOAD_DATE).asDocument().getDateTime("$gt").getValue();
+ // the FileAttribute strategy asks for the file attribute only
+ BsonValue uploadDate = find(document,
GridFsConstants.GRIDFS_FILE_KEY_UPLOAD_DATE);
+ long after = uploadDate != null ?
uploadDate.asDocument().getDateTime("$gt").getValue() : Long.MIN_VALUE;
boolean notMarkedOnly = find(document, "camel-processed") != null;
List<GridFSFile> found = new ArrayList<>();
for (GridFSFile file : stored) {
@@ -283,8 +401,20 @@ class GridFsConsumerTimestampTest {
@SuppressWarnings("unchecked")
private MongoCollection<Document> timestampCollection() {
MongoCollection<Document> timestamps = mock(MongoCollection.class);
+ when(timestamps.countDocuments()).thenAnswer(invocation -> {
+ if (failingTimestampReads.getAndUpdate(n -> Math.max(0, n - 1)) >
0) {
+ onFailedTimestampRead.run();
+ throw new MongoTimeoutException("Timed out while waiting for a
server (test)");
+ }
+ return 0L;
+ });
FindIterable<Document> none = mock(FindIterable.class);
when(timestamps.find(any(Bson.class))).thenReturn(none);
+ when(timestamps.insertOne(any(Document.class))).thenAnswer(invocation
-> {
+ Document document = invocation.getArgument(0);
+ created.add(document.get(GridFsConstants.PERSISTENT_TIMESTAMP_KEY,
Date.class));
+ return null;
+ });
when(timestamps.findOneAndUpdate(any(Bson.class),
any(Bson.class))).thenAnswer(invocation -> {
Bson update = invocation.getArgument(1);
BsonDocument document = toDocument(update);
