allthingssecurity commented on code in PR #27991:
URL: https://github.com/apache/camel/pull/27991#discussion_r4301873493
##########
components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java:
##########
@@ -132,11 +127,54 @@ public void run() {
.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));
+ }
+
+ /**
+ * 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) {
+ try {
+ if (persistsTimestamp && persistentTimestamp == null) {
+ readPersistentTimestamp();
+ }
+ 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) {
+ 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() {
+ 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, new Date());
Review Comment:
Thanks, you are right, the description was wrong there. Restored in
d61890ccea4f as you suggested: `run()` takes `startDate` and passes it to
`poll` and `readPersistentTimestamp`, so a new timestamp document gets the time
the consumer started, also when the first polls cannot reach MongoDB. The
non-persistent `TimeStamp` strategies use the same `startDate`.
New test
`newPersistentTimestampStartsWhenTheConsumerStartsAlthoughTheFirstPollFails`
runs for `PersistentTimestamp` and `PersistentTimestampAndFileAttribute`. The
first timestamp read fails, a file uploaded after that failure (but before the
next poll) must be routed, and the created timestamp must not be later than the
failure. Without the change, both cases fail.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java:
##########
@@ -132,11 +127,54 @@ public void run() {
.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));
+ }
+
+ /**
+ * 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) {
+ try {
+ if (persistsTimestamp && persistentTimestamp == null) {
+ readPersistentTimestamp();
+ }
+ 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) {
+ consecutiveFailures++;
+ if (consecutiveFailures == 1) {
+ getExceptionHandler().handleException("Error polling " +
endpoint + " (will try again on the next poll)", e);
Review Comment:
Done in d61890ccea4f: when the poll fails and `!isRunAllowed()`, it logs at
DEBUG and returns, without going through the `ExceptionHandler` or the failure
counter.
New test `pollFailureWhileTheConsumerStopsIsNotReported`: the consumer is
stopped during a poll, which then throws, and nothing is reported. Without the
change, it gets the exception. camel-mongodb-gridfs tests: 15 run, 0 failures.
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
