nfsantos commented on code in PR #1042:
URL: https://github.com/apache/jackrabbit-oak/pull/1042#discussion_r1316874389
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java:
##########
@@ -215,8 +190,90 @@ private void reportProgress(String id) {
traversalLog.trace(id);
}
- private void downloadRange(DownloadRange range) throws
InterruptedException, TimeoutException {
- BsonDocument findQuery = range.getFindQuery();
+
+ private void downloadWithRetryOnConnectionErrors() throws
InterruptedException, TimeoutException {
+ // If regex filtering is enabled, start by downloading the ancestors
of the path used for filtering.
+ // That is, download "/", "/content", "/content/dam" for a base path
of "/content/dam". These nodes will not be
+ // matched by the regex used in the Mongo query, which assumes a
prefix of "???:/content/dam"
+ Bson childrenFilter = null;
+ String regexBasePath = getPathForRegexFiltering();
+ if (regexBasePath != null) {
+ // Regex path filtering is enabled
+ // Download the ancestors in a separate query. No retrials done on
this query, as it will take only a few
+ // seconds and is done at the start of the job, so if it fails,
the job can be retried without losing much work
+ LOG.info("Using regex filtering with path {}", regexBasePath);
+ Bson ancestorsQuery = ancestorsFilter(regexBasePath);
+ LOG.info("Downloading ancestors of {}", regexBasePath);
+ // Let Mongo decide which index to use for this query, it will
return very few documents
+ FindIterable<BasicDBObject> mongoIterable = dbCollection
+ .withReadPreference(readPreference)
+ .find(ancestorsQuery);
+ download(mongoIterable);
+ // Filter to apply to the main query
+ childrenFilter = childrenFilter(regexBasePath);
+ }
+
+ Instant failuresStartTimestamp = null; // When the last series of
failures started
+ long retryIntervalMs = retryInitialIntervalMillis;
+ int numberOfFailures = 0;
+ boolean downloadCompleted = false;
+ Map<String, Integer> exceptions = new HashMap<>();
+ this.nextLastModified = 0;
+ this.lastIdDownloaded = null;
+ while (!downloadCompleted) {
+ try {
+ if (lastIdDownloaded != null) {
+ LOG.info("Recovering from broken connection, finishing
downloading documents with _modified={}", nextLastModified);
+ downloadRange(new DownloadRange(nextLastModified,
nextLastModified + 1, lastIdDownloaded), childrenFilter);
+ // We have managed to reconnect, reset the failure
timestamp
+ failuresStartTimestamp = null;
+ numberOfFailures = 0;
+ // Continue downloading everything starting from the next
_lastmodified value
+ downloadRange(new DownloadRange(nextLastModified + 1,
Long.MAX_VALUE, null), childrenFilter);
+ } else {
+ downloadRange(new DownloadRange(nextLastModified,
Long.MAX_VALUE, null), childrenFilter);
+ }
+ downloadCompleted = true;
+ } catch (MongoException e) {
+ if (e instanceof MongoInterruptedException || e instanceof
MongoIncompatibleDriverException) {
+ // Non-recoverable exceptions
+ throw e;
+ }
+ if (failuresStartTimestamp == null) {
+ failuresStartTimestamp =
Instant.now().truncatedTo(ChronoUnit.SECONDS);
+ }
+ LOG.warn("Connection error downloading from MongoDB.", e);
+ long secondsSinceStartOfFailures =
Duration.between(failuresStartTimestamp, Instant.now()).toSeconds();
+ if (secondsSinceStartOfFailures > retryDuringSeconds) {
+ // Give up. Get a string of all exceptions that were thrown
+ StringBuilder summary = new StringBuilder();
+ for (Map.Entry<String, Integer> entry :
exceptions.entrySet()) {
+
summary.append("\n\t").append(entry.getValue()).append("x:
").append(entry.getKey());
+ }
+ throw new RetryException(retryDuringSeconds,
summary.toString(), e);
+ } else {
+ numberOfFailures++;
+ LOG.warn("Retrying download in {} ms; number of times
failed: {}; current series of failures started at: {} ({} seconds ago)",
+ retryIntervalMs, numberOfFailures,
failuresStartTimestamp, secondsSinceStartOfFailures);
+ exceptions.compute(e.getClass().getSimpleName() + " - " +
e.getMessage(),
+ (key, val) -> val == null ? 1 : val + 1
+ );
+ try {
+ Thread.sleep(retryIntervalMs);
+ } catch (InterruptedException ignore) {
+ }
Review Comment:
Yes, you are right. The method with this code already declares throwing
`InterruptedException` and the caller is handling that exception explicitly, so
the catch clause is not needed here. I removed it.
--
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: dev-unsubscr...@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
