JVMCBC-1477 (Further) Improve config loading backoff for bad credentials

Motivation
----------
The previous change for JVMCBC-1477 increased the retry delay
for bad credentials from 1 millisecond to 500 milliseconds.
500ms is probably still too short, as it produces two
entries per second in the server's http_access.log.

Modifications
-------------
Increase the fixed "short delay" from 1ms to 10ms.

Use exponential backoff (with the default .5 jitter factor)
for the "long delay" exceptions like bad credentials.

Implement this by chaining `retryWhen` operators.

Consequently, the BucketOpenRetriedEvent is now published
when the retry is _scheduled_, not when it's executed.
Also, the retry delay is not available to include in the event.
These minor changes are acceptable, since the event's
Javadoc says it exists "for debugging reasons".

Change-Id: Iee603ed6aaab105240fd8982963087bda8fd7cc7
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/207419
Reviewed-by: Michael Reiche <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: David Nault <[email protected]>

Author: dnault

core-io/src/main/java/com/couchbase/client/core/config/DefaultConfigurationProvider.java

@@ -74,6 +74,7 @@
 
 import javax.naming.NamingException;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -83,6 +84,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -962,33 +964,49 @@ private Mono<ProposedBucketConfigContext> fetchBucketConfigs(final String name,
 
           return loadBucketConfigForSeed(identifier, mappedKvPort, mappedManagerPort, name, alternateAddress);
         })
-        .retryWhen(
-          Retry.from(companion -> companion.flatMap(rs -> {
-            final Throwable f = rs.failure();
-            if (shutdown.get()) {
-              return Mono.error(new AlreadyShutdownException());
-            }
-            if (f instanceof UnsupportedConfigMechanismException) {
-              return Mono.error(Exceptions.propagate(f));
-            }
-
-            boolean bucketNotFound = f instanceof BucketNotFoundDuringLoadException;
-            boolean bucketNotReady = f instanceof BucketNotReadyDuringLoadException;
-            boolean noAccess = f instanceof NoAccessDuringConfigLoadException;
-
-            // For some, wait a bit longer; retry the rest quickly.
-            Duration delay = bucketNotFound || bucketNotReady || noAccess
-              ? Duration.ofMillis(500)
-              : Duration.ofMillis(1);
-            eventBus.publish(new BucketOpenRetriedEvent(name, delay, core.context(), f));
-            return Mono
-              .just(rs.totalRetries())
-              .delayElement(delay, core.context().environment().scheduler());
-          })))
+        // Exponential backoff for certain errors.
+        .retryWhen(Retry
+          .backoff(Long.MAX_VALUE, Duration.ofMillis(500))
+          .maxBackoff(Duration.ofSeconds(10))
+          .filter(bucketConfigLoadRetryFilter(name, t -> isInstanceOfAnyOf(t,
+            BucketNotFoundDuringLoadException.class,
+            BucketNotReadyDuringLoadException.class,
+            NoAccessDuringConfigLoadException.class
+          )))
+        )
+        // Short fixed delay for the others we want to retry.
+        .retryWhen(Retry
+          .fixedDelay(Long.MAX_VALUE, Duration.ofMillis(10))
+          .filter(bucketConfigLoadRetryFilter(name, t -> !(t instanceof UnsupportedConfigMechanismException)))
+        )
       )
       .next();
   }
 
+  /**
+   * Wraps a retry filter with common logic that checks for shutdown and publishes events.
+   */
+  private Predicate<? super Throwable> bucketConfigLoadRetryFilter(
+    String bucketName,
+    Predicate<? super Throwable> errorFilter
+  ) {
+    return t -> {
+      if (shutdown.get()) {
+        throw new AlreadyShutdownException();
+      }
+
+      boolean retry = errorFilter.test(t);
+      if (retry) {
+        eventBus.publish(new BucketOpenRetriedEvent(bucketName, Duration.ZERO, core.context(), t));
+      }
+      return retry;
+    };
+  }
+
+  private static boolean isInstanceOfAnyOf(Object o, Class<?>... candidates) {
+    return Arrays.stream(candidates).anyMatch(it -> it.isInstance(o));
+  }
+
   private Mono<ProposedGlobalConfigContext> fetchGlobalConfigs(final Set<SeedNode> seedNodes, final boolean tls,
                                                                boolean allowStaleSeeds, boolean retryTimeouts) {
     final AtomicBoolean hasErrored = new AtomicBoolean();