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 5bef69484652 CAMEL-25607: camel-mongodb-gridfs - keep the timestamp of
the consumer between the polls (#27794)
5bef69484652 is described below
commit 5bef6948465255c2205a1058df8e8c9701363d53
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 15 11:19:01 2026 +0530
CAMEL-25607: camel-mongodb-gridfs - keep the timestamp of the consumer
between the polls (#27794)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
components/camel-mongodb-gridfs/pom.xml | 6 +
.../component/mongodb/gridfs/GridFsConsumer.java | 26 +-
.../gridfs/GridFsConsumerTimestampTest.java | 297 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +
4 files changed, 333 insertions(+), 8 deletions(-)
diff --git a/components/camel-mongodb-gridfs/pom.xml
b/components/camel-mongodb-gridfs/pom.xml
index 63fb0dda8c80..14b7707909f6 100644
--- a/components/camel-mongodb-gridfs/pom.xml
+++ b/components/camel-mongodb-gridfs/pom.xml
@@ -79,5 +79,11 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
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 06223f1c63ff..40afc3d4f142 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
@@ -51,6 +51,13 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
private final GridFsEndpoint endpoint;
private volatile ScheduledExecutorService taskExecutor;
private volatile ExecutorService executor;
+ // the polls ask for the files uploaded after this date. With the
TimeStamp and PersistentTimestamp strategies it
+ // is the upload date of the newest file that was routed, kept between the
polls. The strategies with a file
+ // attribute keep the start date, because the attribute already makes sure
that a file is routed once, and a file
+ // whose upload date is older than one that was routed already (another
uploader, clock skew) is still found.
+ // 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;
public GridFsConsumer(GridFsEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -82,7 +89,7 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
@Override
public void run() {
- Date fromDate = null;
+ fromDate = null;
QueryStrategy queryStrategy = endpoint.getQueryStrategy();
boolean usesTimestamp = queryStrategy != QueryStrategy.FileAttribute;
@@ -126,22 +133,22 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
.build();
MongoCollection<Document> finalPtsCollection = ptsCollection;
- Date finalFromDate = fromDate;
Document finalPersistentTimestamp = persistentTimestamp;
task.run(endpoint.getCamelContext(),
- () -> processCollection(finalFromDate, usesTimestamp,
persistsTimestamp, usesAttribute, finalPtsCollection,
+ () -> processCollection(usesTimestamp, persistsTimestamp,
usesAttribute, finalPtsCollection,
finalPersistentTimestamp));
}
private boolean processCollection(
- Date fromDate, boolean usesTimestamp, boolean persistsTimestamp,
boolean usesAttribute,
+ boolean usesTimestamp, boolean persistsTimestamp, boolean
usesAttribute,
final MongoCollection<Document> ptsCollection, final Document
persistentTimestamp) {
if (!isStarted()) {
return false;
}
- try (MongoCursor<GridFSFile> cursor =
getGridFSFileMongoCursor(fromDate, usesTimestamp, usesAttribute)) {
+ Date pollFromDate = fromDate;
+ try (MongoCursor<GridFSFile> cursor =
getGridFSFileMongoCursor(pollFromDate, usesTimestamp, usesAttribute)) {
boolean dateModified = false;
while (cursor.hasNext() && isStarted()) {
@@ -177,9 +184,12 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
Bson update =
Updates.set(endpoint.getFileAttributeName(), GRIDFS_FILE_ATTRIBUTE_DONE);
endpoint.getFilesCollection().findOneAndUpdate(eq("_id", fOrig.getId()),
update);
}
- if (usesTimestamp &&
file.getUploadDate().compareTo(fromDate) > 0) {
- fromDate = file.getUploadDate();
+ if (usesTimestamp &&
file.getUploadDate().compareTo(pollFromDate) > 0) {
+ pollFromDate = file.getUploadDate();
dateModified = true;
+ if (!usesAttribute) {
+ fromDate = pollFromDate;
+ }
}
} catch (Exception e) {
// ignore
@@ -188,7 +198,7 @@ public class GridFsConsumer extends DefaultConsumer
implements Runnable {
}
if (persistsTimestamp && dateModified) {
- Bson update = Updates.set(PERSISTENT_TIMESTAMP_KEY, fromDate);
+ Bson update = Updates.set(PERSISTENT_TIMESTAMP_KEY,
pollFromDate);
ptsCollection.findOneAndUpdate(eq("_id",
persistentTimestamp.getObjectId("_id")), update);
}
}
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
new file mode 100644
index 000000000000..47a214d76dfc
--- /dev/null
+++
b/components/camel-mongodb-gridfs/src/test/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumerTimestampTest.java
@@ -0,0 +1,297 @@
+/*
+ * 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.gridfs;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.mongodb.MongoClientSettings;
+import com.mongodb.client.FindIterable;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoCursor;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.gridfs.GridFSBucket;
+import com.mongodb.client.gridfs.GridFSDownloadStream;
+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.Processor;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.bson.BsonArray;
+import org.bson.BsonDocument;
+import org.bson.BsonObjectId;
+import org.bson.BsonValue;
+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 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.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The consumer with a timestamp query strategy against a fake bucket that
answers the upload date query of each poll
+ * the way MongoDB does: a file that was routed must not be routed again by
the following polls. With a file attribute,
+ * the fake files collection marks a file as the consumer asks, and the bucket
no longer returns a marked file.
+ */
+class GridFsConsumerTimestampTest {
+
+ private static final long TIMEOUT_SECONDS = 20;
+
+ private final List<GridFSFile> stored = new CopyOnWriteArrayList<>();
+ private final List<String> routed = new CopyOnWriteArrayList<>();
+ 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();
+ private final GridFSBucket bucket = mock(GridFSBucket.class);
+ private final MongoDatabase database = mock(MongoDatabase.class);
+ @SuppressWarnings("unchecked")
+ private final MongoCollection<GridFSFile> files =
mock(MongoCollection.class);
+
+ private volatile CountDownLatch polls = new CountDownLatch(0);
+ private CamelContext context;
+ private GridFsConsumer consumer;
+
+ @BeforeEach
+ void setUp() {
+ context = new DefaultCamelContext();
+ context.start();
+
+ when(bucket.find(any(Bson.class))).thenAnswer(invocation -> {
+ List<GridFSFile> found =
filesUploadedAfter(invocation.getArgument(0));
+ polls.countDown();
+ return iterableOf(found);
+ });
+ when(bucket.openDownloadStream(anyString())).thenAnswer(invocation -> {
+ String name = invocation.getArgument(0);
+ GridFSDownloadStream stream = mock(GridFSDownloadStream.class);
+ when(stream.getGridFSFile())
+ .thenReturn(stored.stream().filter(f ->
f.getFilename().equals(name)).findFirst().orElseThrow());
+ return stream;
+ });
+ // the consumer marks a file before it routes it: only a file that is
not marked yet is returned
+ when(files.findOneAndUpdate(any(Bson.class), any(Bson.class),
any(FindOneAndUpdateOptions.class)))
+ .thenAnswer(invocation -> {
+ BsonValue id = find(toDocument(invocation.getArgument(0)),
"_id");
+ return marked.add(id)
+ ? stored.stream().filter(f ->
f.getId().equals(id)).findFirst().orElseThrow() : null;
+ });
+ MongoCollection<Document> timestamps = timestampCollection();
+
when(database.getCollection("camel-timestamps")).thenReturn(timestamps);
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (consumer != null) {
+ consumer.stop();
+ }
+ context.stop();
+ }
+
+ @Test
+ void timestampStrategyRoutesAFileOnce() throws Exception {
+ startConsumer(QueryStrategy.TimeStamp);
+
+ stored.add(file("first", 60));
+ awaitFullPolls(3);
+
+ assertEquals(List.of("first"), routed, "three polls saw the file, it
must be routed by the first of them only");
+ }
+
+ @Test
+ void timestampStrategyRoutesALaterFileOnce() throws Exception {
+ startConsumer(QueryStrategy.TimeStamp);
+
+ stored.add(file("first", 60));
+ awaitFullPolls(2);
+ stored.add(file("second", 120));
+ awaitFullPolls(3);
+
+ assertEquals(List.of("first", "second"), routed);
+ }
+
+ @Test
+ void persistentTimestampStrategyRoutesAFileOnce() throws Exception {
+ startConsumer(QueryStrategy.PersistentTimestamp);
+
+ GridFSFile first = file("first", 60);
+ stored.add(first);
+ awaitFullPolls(3);
+
+ assertEquals(List.of("first"), routed, "three polls saw the file, it
must be routed by the first of them only");
+ assertEquals(List.of(first.getUploadDate()), persisted, "the timestamp
is stored when it moves, not at every poll");
+ }
+
+ @Test
+ void timestampStrategySkipsAFileWithAnOlderUploadDateThanARoutedFile()
throws Exception {
+ startConsumer(QueryStrategy.TimeStamp);
+
+ stored.add(file("newer", 120));
+ awaitFullPolls(2);
+ // visible only now, but uploaded before the routed file (another
uploader, or a client with another clock)
+ stored.add(file("older", 60));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("newer"), routed, "the strategy only asks for
files uploaded after the newest routed file");
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = QueryStrategy.class, names = {
"TimeStampAndFileAttribute", "PersistentTimestampAndFileAttribute" })
+ void
strategyWithFileAttributeRoutesAFileWithAnOlderUploadDateThanARoutedFile(QueryStrategy
strategy)
+ throws Exception {
+ startConsumer(strategy);
+
+ stored.add(file("newer", 120));
+ awaitFullPolls(2);
+ // visible only now, but uploaded before the routed file (another
uploader, or a client with another clock)
+ stored.add(file("older", 60));
+ awaitFullPolls(2);
+
+ assertEquals(List.of("newer", "older"), routed,
+ "the file attribute makes sure a file is routed once, the
query keeps the start date");
+ }
+
+ private void startConsumer(QueryStrategy strategy) {
+ GridFsComponent component = new GridFsComponent();
+ component.setCamelContext(context);
+ GridFsEndpoint endpoint = new GridFsEndpoint("mongodb-gridfs:fake",
component) {
+ @Override
+ public GridFSBucket getGridFsBucket() {
+ return bucket;
+ }
+
+ @Override
+ public MongoDatabase getDB() {
+ return database;
+ }
+
+ @Override
+ public MongoCollection<GridFSFile> getFilesCollection() {
+ return files;
+ }
+ };
+ endpoint.setCamelContext(context);
+ endpoint.setQueryStrategy(strategy);
+ endpoint.setInitialDelay(1);
+ endpoint.setDelay(1);
+
+ Processor processor = exchange ->
routed.add(exchange.getMessage().getBody(GridFSDownloadStream.class)
+ .getGridFSFile().getFilename());
+ consumer = new GridFsConsumer(endpoint, processor);
+ consumer.start();
+ }
+
+ /**
+ * Waits until the consumer has started that many polls and one more, so
that many polls have run to their end: the
+ * consumer polls on one thread.
+ */
+ private void awaitFullPolls(int count) throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(count + 1);
+ polls = latch;
+ assertTrue(latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "the
consumer did not poll " + (count + 1) + " times");
+ }
+
+ /**
+ * A file uploaded the given number of seconds after the test (and so the
consumer) started.
+ */
+ private static GridFSFile file(String name, int secondsFromNow) {
+ Date uploadDate = new Date(System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(secondsFromNow));
+ return new GridFSFile(new BsonObjectId(), name, 1, 1, uploadDate,
null);
+ }
+
+ /**
+ * 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>.
+ */
+ private List<GridFSFile> filesUploadedAfter(Bson query) {
+ BsonDocument document = toDocument(query);
+ long after = find(document,
GridFsConstants.GRIDFS_FILE_KEY_UPLOAD_DATE).asDocument().getDateTime("$gt").getValue();
+ boolean notMarkedOnly = find(document, "camel-processed") != null;
+ List<GridFSFile> found = new ArrayList<>();
+ for (GridFSFile file : stored) {
+ if (file.getUploadDate().getTime() > after && !(notMarkedOnly &&
marked.contains(file.getId()))) {
+ found.add(file);
+ }
+ }
+ return found;
+ }
+
+ private static BsonDocument toDocument(Bson bson) {
+ return bson.toBsonDocument(BsonDocument.class,
MongoClientSettings.getDefaultCodecRegistry());
+ }
+
+ /**
+ * The value of the given key in a filter, or in one of the filters of an
<tt>$and</tt>.
+ */
+ private static BsonValue find(BsonDocument filter, String key) {
+ if (filter.containsKey(key)) {
+ return filter.get(key);
+ }
+ BsonArray and = filter.getArray("$and", new BsonArray());
+ for (BsonValue part : and) {
+ BsonValue value = find(part.asDocument(), key);
+ if (value != null) {
+ return value;
+ }
+ }
+ return null;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static GridFSFindIterable iterableOf(List<GridFSFile> files) {
+ Iterator<GridFSFile> iterator = files.iterator();
+ MongoCursor<GridFSFile> cursor = mock(MongoCursor.class);
+ when(cursor.hasNext()).thenAnswer(invocation -> iterator.hasNext());
+ when(cursor.next()).thenAnswer(invocation -> iterator.next());
+ GridFSFindIterable iterable = mock(GridFSFindIterable.class);
+ when(iterable.cursor()).thenReturn(cursor);
+ return iterable;
+ }
+
+ /**
+ * The collection of the persistent timestamp: empty at first, records the
dates the consumer stores in it.
+ */
+ @SuppressWarnings("unchecked")
+ private MongoCollection<Document> timestampCollection() {
+ MongoCollection<Document> timestamps = mock(MongoCollection.class);
+ FindIterable<Document> none = mock(FindIterable.class);
+ when(timestamps.find(any(Bson.class))).thenReturn(none);
+ when(timestamps.findOneAndUpdate(any(Bson.class),
any(Bson.class))).thenAnswer(invocation -> {
+ Bson update = invocation.getArgument(1);
+ BsonDocument document = toDocument(update);
+ persisted.add(new Date(
+
document.getDocument("$set").getDateTime(GridFsConstants.PERSISTENT_TIMESTAMP_KEY).getValue()));
+ return null;
+ });
+ return timestamps;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index fdb8d76a2559..68e31b6ceb25 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -711,3 +711,15 @@ project in the same way.
A name in camel case that starts with a lower-case letter, such as the
datasource `myOrders` or `my_DataSource`, still
looks like an option name and is written in dash case as a whole, underscores
included (`my-orders`,
`my-data-source`). Quote it to keep it:
`quarkus.datasource."myOrders".db-kind`.
+
+=== camel-mongodb-gridfs - the consumer routes a file once
+
+With the `TimeStamp` (default) and `PersistentTimestamp` query strategies, the
`mongodb-gridfs` consumer routed
+every file uploaded since the consumer started again at each poll (every
`delay`, 500 ms by default). The upload
+date of the newest routed file is now kept between the polls, so a file is
routed once. A route that relied on
+receiving the same file repeatedly now receives it once.
+These two strategies now skip a file whose upload date is older than that of a
file already routed (as they did
+before Camel 3.13), for example a file from a second uploader, or from a
client with another clock, that becomes
+visible later: GridFS sets the upload date on the client. Use
`TimeStampAndFileAttribute` or
+`PersistentTimestampAndFileAttribute` if such files must be routed: these
strategies keep the start date in their
+query, as before, and mark each file so that it is routed once.
