Introduce the RetryPolicy abstraction (#2004)

JAVA-6229

Author: stIncMale

driver-core/src/main/com/mongodb/MongoClientSettings.java

@@ -542,7 +542,8 @@ public Builder retryReads(final boolean retryReads) {
          *                 The attempts failed due to retryable overload errors are retried only if
          *                 {@link #retryWrites(boolean)} is {@code true} and {@link #retryReads(boolean)} is {@code true}.
          *                 <p>
-         *                 The command kind is unknown when a command is executed via the {@code MongoDatabase.runCommand} operation.
+         *                 The command kind is unknown if and only if a command is executed via
+         *                 the {@code MongoDatabase.runCommand} operation.
          *             </td>
          *         </tr>
          *     </tbody>
@@ -772,14 +773,11 @@ public Builder autoEncryptionSettings(@Nullable final AutoEncryptionSettings aut
 
         /**
          * Sets the context provider
-         *
          * <p>
          * When used with the synchronous driver, this must be an instance of {@code com.mongodb.client.SynchronousContextProvider}.
          * When used with the reactive streams driver, this must be an instance of
          * {@code com.mongodb.reactivestreams.client.ReactiveContextProvider}.
          *
-         * </p>
-         *
          * @param contextProvider the context provider
          * @return this
          * @since 4.4

driver-core/src/main/com/mongodb/internal/TimeoutContext.java

@@ -40,6 +40,7 @@
  * <p>The context for handling timeouts in relation to the Client Side Operation Timeout specification.</p>
  */
 public class TimeoutContext {
+    public static final String DEFAULT_TIMEOUT_MESSAGE = "Operation exceeded the timeout limit.";
     private static final int NO_ROUND_TRIP_TIME_MS = 0;
     private final TimeoutSettings timeoutSettings;
     /**
@@ -64,11 +65,11 @@ public static <T> T throwMongoTimeoutException(final String message) {
         throw new MongoOperationTimeoutException(message);
     }
     public static <T> T throwMongoTimeoutException() {
-        throw new MongoOperationTimeoutException("The operation exceeded the timeout limit.");
+        throw new MongoOperationTimeoutException(DEFAULT_TIMEOUT_MESSAGE);
     }
 
     public static MongoOperationTimeoutException createMongoTimeoutException(final Throwable cause) {
-        return createMongoTimeoutException("Operation exceeded the timeout limit: " + cause.getMessage(), cause);
+        return createMongoTimeoutException(DEFAULT_TIMEOUT_MESSAGE, cause);
     }
 
     public static MongoOperationTimeoutException createMongoTimeoutException(final String message, @Nullable final Throwable cause) {
@@ -185,7 +186,7 @@ public long timeoutOrAlternative(final long alternativeTimeoutMS) {
             return timeout.call(MILLISECONDS,
                     () -> 0L,
                     (ms) -> ms,
-                    () -> throwMongoTimeoutException("The operation exceeded the timeout limit."));
+                    () -> throwMongoTimeoutException());
         }
     }
 
@@ -229,7 +230,7 @@ public int getConnectTimeoutMs() {
         return Math.toIntExact(Timeout.nullAsInfinite(timeout).call(MILLISECONDS,
                 () -> connectTimeoutMS,
                 (ms) -> connectTimeoutMS == 0 ? ms : Math.min(ms, connectTimeoutMS),
-                () -> throwMongoTimeoutException("The operation exceeded the timeout limit.")));
+                () -> throwMongoTimeoutException()));
     }
 
     /**
@@ -250,13 +251,11 @@ public TimeoutContext withMaxTimeAsMaxAwaitTimeOverride() {
      * The override will be provided as the remaining value in
      * {@link #runMaxTimeMS}, where 0 is ignored. This is useful for setting timeout
      * in {@link CommandMessage} as an extra element before we send it to the server.
-     *
      * <p>
-     * NOTE: Suitable for static user-defined values only (i.e MaxAwaitTimeMS),
+     * Suitable for static user-defined values only (i.e. {@code MaxAwaitTimeMS}),
      * not for running timeouts that adjust dynamically (CSOT).
-     *
+     * <p>
      * If remaining CSOT timeout is less than this static timeout, then CSOT timeout will be used.
-     *
      */
     public TimeoutContext withMaxTimeOverride(final long maxTimeMS) {
         return new TimeoutContext(
@@ -480,10 +479,10 @@ private void runMinTimeout(final LongConsumer onRemaining, final long fixedMs) {
             timeout.run(MILLISECONDS, () -> {
                         onRemaining.accept(fixedMs);
                     },
-                    (renamingMs) -> {
-                        onRemaining.accept(Math.min(renamingMs, fixedMs));
+                    (remainingMs) -> {
+                        onRemaining.accept(Math.min(remainingMs, fixedMs));
                     }, () -> {
-                        throwMongoTimeoutException("The operation exceeded the timeout limit.");
+                        throwMongoTimeoutException();
                     });
         } else {
             onRemaining.accept(fixedMs);

driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java

@@ -17,8 +17,8 @@
 package com.mongodb.internal.async;
 
 import com.mongodb.internal.async.function.AsyncCallbackLoop;
-import com.mongodb.internal.async.function.LoopState;
-import com.mongodb.internal.async.function.RetryState;
+import com.mongodb.internal.async.function.LoopControl;
+import com.mongodb.internal.async.function.RetryControl;
 import com.mongodb.internal.async.function.RetryingAsyncCallbackSupplier;
 
 import java.util.function.BooleanSupplier;
@@ -233,9 +233,7 @@ default <R> AsyncSupplier<R> thenSupply(final AsyncSupplier<R> supplier) {
     default AsyncRunnable thenRunRetryingWhile(final AsyncRunnable runnable, final Predicate<Throwable> shouldRetry) {
         return thenRun(callback -> {
             new RetryingAsyncCallbackSupplier<Void>(
-                    new RetryState(),
-                    (previouslyChosenFailure, lastAttemptFailure) -> lastAttemptFailure,
-                    (rs, lastAttemptFailure) -> shouldRetry.test(lastAttemptFailure),
+                    new RetryControl<>(new SimpleRetryPolicy(shouldRetry)),
                     // `finish` is required here instead of `unsafeFinish`
                     // because only `finish` meets the contract of
                     // `AsyncCallbackSupplier.get`, which we implement here
@@ -255,10 +253,10 @@ default AsyncRunnable thenRunRetryingWhile(final AsyncRunnable runnable, final P
      */
     default AsyncRunnable thenRunWhileLoop(final BooleanSupplier whileCheck, final AsyncRunnable loopBodyRunnable) {
         return thenRun(finalCallback -> {
-            LoopState loopState = new LoopState();
-            new AsyncCallbackLoop(loopState, iterationCallback -> {
+            LoopControl loopControl = new LoopControl();
+            new AsyncCallbackLoop(loopControl, iterationCallback -> {
 
-                if (loopState.breakAndCompleteIf(() -> !whileCheck.getAsBoolean(), iterationCallback)) {
+                if (loopControl.breakAndCompleteIf(() -> !whileCheck.getAsBoolean(), iterationCallback)) {
                     return;
                 }
                 loopBodyRunnable.finish((result, t) -> {
@@ -284,15 +282,15 @@ default AsyncRunnable thenRunWhileLoop(final BooleanSupplier whileCheck, final A
      */
     default AsyncRunnable thenRunDoWhileLoop(final AsyncRunnable loopBodyRunnable, final BooleanSupplier whileCheck) {
         return thenRun(finalCallback -> {
-            LoopState loopState = new LoopState();
-            new AsyncCallbackLoop(loopState, iterationCallback -> {
+            LoopControl loopControl = new LoopControl();
+            new AsyncCallbackLoop(loopControl, iterationCallback -> {
 
                 loopBodyRunnable.finish((result, t) -> {
                     if (t != null) {
                         iterationCallback.completeExceptionally(t);
                         return;
                     }
-                    if (loopState.breakAndCompleteIf(() -> !whileCheck.getAsBoolean(), iterationCallback)) {
+                    if (loopControl.breakAndCompleteIf(() -> !whileCheck.getAsBoolean(), iterationCallback)) {
                         return;
                     }
                     iterationCallback.complete(iterationCallback);

driver-core/src/main/com/mongodb/internal/async/SimpleRetryPolicy.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed 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 com.mongodb.internal.async;
+
+import com.mongodb.internal.async.function.RetryContext;
+import com.mongodb.internal.async.function.RetryPolicy;
+import com.mongodb.internal.async.function.RetryPolicy.Decision.RetryAttemptInfo;
+
+import java.util.function.Predicate;
+
+final class SimpleRetryPolicy implements RetryPolicy {
+    private final Predicate<Throwable> shouldRetry;
+
+    SimpleRetryPolicy(final Predicate<Throwable> shouldRetry) {
+        this.shouldRetry = shouldRetry;
+    }
+
+    @Override
+    public Decision onAttemptFailure(final RetryContext retryContext, final Throwable attemptFailedResult) {
+        return new Decision(attemptFailedResult, shouldRetry.test(attemptFailedResult) ? new RetryAttemptInfo() : null);
+    }
+}

driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackFunction.java

@@ -35,8 +35,8 @@
  * <a href="https://docs.oracle.com/javase/specs/jls/se8/html/jls-14.html#jls-14.1">"normal" and "abrupt completion"</a>
  * are used as they defined by the Java Language Specification, while the terms "successful" and "failed completion" are used to refer to a
  * situation when the function produces either a successful or a failed result respectively.
- *
- * <p>This class is not part of the public API and may be removed or changed at any time</p>
+ * <p>
+ * This class is not part of the public API and may be removed or changed at any time.
  *
  * @param <P> The type of the first parameter to the function.
  * @param <R> The type of successful result. A failed result is of the {@link Throwable} type

driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackLoop.java

@@ -29,26 +29,26 @@
  * This class emulates the <a href="https://docs.oracle.com/javase/specs/jls/se8/html/jls-14.html#jls-14.12">{@code while(true)}</a>
  * statement.
  * <p>
- * The original function may additionally observe or control looping via {@link LoopState}.
- * Looping continues until either of the following happens:
+ * The original function may additionally observe or control the loop via {@link LoopControl}.
+ * The loop continues until either of the following happens:
  * <ul>
  *     <li>the original function fails as specified by {@link AsyncCallbackFunction};</li>
- *     <li>the original function calls {@link LoopState#breakAndCompleteIf(Supplier, SingleResultCallback)}.</li>
+ *     <li>the original function calls {@link LoopControl#breakAndCompleteIf(Supplier, SingleResultCallback)}.</li>
  * </ul>
- *
- * <p>This class is not part of the public API and may be removed or changed at any time</p>
+ * <p>
+ * This class is not part of the public API and may be removed or changed at any time.
  */
 @NotThreadSafe
 public final class AsyncCallbackLoop implements AsyncCallbackRunnable {
-    private final LoopState state;
+    private final LoopControl control;
     private final AsyncCallbackRunnable body;
 
     /**
-     * @param state The {@link LoopState} to be deemed as initial for the purpose of the new {@link AsyncCallbackLoop}.
+     * @param control The {@link LoopControl} to control the new {@link AsyncCallbackLoop}.
      * @param body The body of the loop.
      */
-    public AsyncCallbackLoop(final LoopState state, final AsyncCallbackRunnable body) {
-        this.state = state;
+    public AsyncCallbackLoop(final LoopControl control, final AsyncCallbackRunnable body) {
+        this.control = control;
         this.body = body;
     }
 
@@ -77,7 +77,7 @@ public void onResult(@Nullable final Void result, @Nullable final Throwable t) {
             } else {
                 boolean continueLooping;
                 try {
-                    continueLooping = state.advance();
+                    continueLooping = control.advance();
                 } catch (Throwable e) {
                     wrapped.onResult(null, e);
                     return;

driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackRunnable.java

@@ -20,8 +20,8 @@
 /**
  * An {@linkplain AsyncCallbackFunction asynchronous callback-based function} of no parameters and no successful result.
  * This class is a callback-based counterpart of {@link Runnable}.
- *
- * <p>This class is not part of the public API and may be removed or changed at any time</p>
+ * <p>
+ * This class is not part of the public API and may be removed or changed at any time.
  *
  * @see AsyncCallbackFunction
  */

driver-core/src/main/com/mongodb/internal/async/function/LoopControl.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed 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 com.mongodb.internal.async.function;
+
+import com.mongodb.annotations.NotThreadSafe;
+import com.mongodb.internal.async.MutableValue;
+import com.mongodb.internal.async.SingleResultCallback;
+
+import java.util.function.Supplier;
+
+import static com.mongodb.assertions.Assertions.assertFalse;
+
+/**
+ * A stateful controller of a loop that can be used to control it, for example,
+ * to {@linkplain #breakAndCompleteIf(Supplier, SingleResultCallback) break} it.
+ * {@linkplain MutableValue} may be used by the loop to preserve state between iterations.
+ * <p>
+ * This class is not part of the public API and may be removed or changed at any time.
+ *
+ * @see AsyncCallbackLoop
+ */
+@NotThreadSafe
+public final class LoopControl {
+    private int iteration;
+    private boolean lastIteration;
+
+    public LoopControl() {
+        iteration = 0;
+    }
+
+    /**
+     * Advances this {@link LoopControl} such that it represents the state of the immediate next iteration, if any.
+     * Must not be called before the {@linkplain #isFirstIteration() first iteration}, must be called before each subsequent iteration.
+     *
+     * @return {@code true} if another iteration must be executed;
+     * otherwise the loop was {@link #isLastIteration() broken} and {@code false} is returned.
+     */
+    boolean advance() {
+        if (lastIteration) {
+            return false;
+        } else {
+            iteration++;
+            return true;
+        }
+    }
+
+    /**
+     * Returns {@code true} iff the current iteration is the first one.
+     *
+     * @see #iteration()
+     */
+    boolean isFirstIteration() {
+        return iteration == 0;
+    }
+
+    /**
+     * Returns {@code true} iff {@link #breakAndCompleteIf(Supplier, SingleResultCallback)} / {@link #markAsLastIteration()} was called.
+     */
+    boolean isLastIteration() {
+        return lastIteration;
+    }
+
+    /**
+     * A 0-based iteration number.
+     */
+    int iteration() {
+        return iteration;
+    }
+
+    /**
+     * This method emulates executing the <a href="https://docs.oracle.com/javase/specs/jls/se8/html/jls-14.html#jls-14.15">{@code break}</a> statement
+     * in callback-based code. If {@code true} is returned, the caller must complete the current attempt.
+     * <p>
+     * Must not be called after breaking the loop.
+     *
+     * @param predicate {@code true} iff the loop needs to be broken.
+     * <ul>
+     *  <li>
+     *  If the {@code predicate} completes abruptly, this method completes the {@code callback} with the same exception but does not break the loop;</li>
+     *  <li>
+     *  if the {@code predicate} is {@code true}, then this method breaks the retry loop;</li>
+     *  <li>
+     *  if the {@code predicate} is {@code false}, then this method does nothing.
+     * </ul>
+     * @return {@code true} iff the {@code callback} was completed, which happens iff any of the following is true:
+     * <ul>
+     *     <li>the {@code predicate} completed abruptly;</li>
+     *     <li>this method broke the loop.</li>
+     * </ul>
+     *
+     * @see #isLastIteration()
+     */
+    public boolean breakAndCompleteIf(final Supplier<Boolean> predicate, final SingleResultCallback<?> callback) {
+        assertFalse(lastIteration);
+        try {
+            lastIteration = predicate.get();
+        } catch (Throwable predicateException) {
+            callback.onResult(null, predicateException);
+            return true;
+        }
+        if (lastIteration) {
+            callback.onResult(null, null);
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * This method is similar to {@link #breakAndCompleteIf(Supplier, SingleResultCallback)}.
+     * The difference is that it allows the current iteration to continue, yet no more iterations will happen.
+     *
+     * @see #isLastIteration()
+     */
+    void markAsLastIteration() {
+        assertFalse(lastIteration);
+        lastIteration = true;
+    }
+
+    @Override
+    public String toString() {
+        return "LoopControl{"
+                + "iteration=" + iteration
+                + ", lastIteration=" + lastIteration
+                + '}';
+    }
+}