davsclaus commented on code in PR #27991:
URL: https://github.com/apache/camel/pull/27991#discussion_r4301786363
##########
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:
Small behaviour change for a consumer that starts with no timestamp document
yet. Before, the new document got the time `run()` ran, which is when the
consumer started. Now it gets the time of the first poll that works, which is
at least `initialDelay` (1 s by default) later. A file uploaded in that window
has `uploadDate` < the stored value, so `PersistentTimestamp` never routes it.
With `PersistentTimestampAndFileAttribute` the start date is stored for good.
Taking the date in `run()` and passing it down keeps the old value. It also
covers files that other clients upload while this consumer cannot reach MongoDB
at startup:
```java
// run()
Date startDate = new Date();
...
task.run(endpoint.getCamelContext(), () -> poll(usesTimestamp,
persistsTimestamp, usesAttribute, startDate));
// readPersistentTimestamp(Date startDate)
timestamp.put(PERSISTENT_TIMESTAMP_KEY, startDate);
```
##########
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:
Nit: when a poll fails because the consumer is stopping (for example the
`MongoClient` is closed during shutdown), this logs a WARN with a stack trace
that says it will try again, but no poll follows. You could skip the report in
that case:
```java
if (!isRunAllowed()) {
LOG.debug("Error polling {} while stopping: {}", endpoint,
e.getMessage());
return false;
}
```
--
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]
