Refactoring for backpressure (#1952)

Refactoring
- naming, log messages, internal docs
- remove unused `binding` parameters from `MixedBulkWriteOperation`
-`ClientBulkWriteOperation`:
-- Document the `doWithRetriesDisabled*` methods, leave `TODO-JAVA-5956` to fix them.
-- Add `ClientBulkWriteCommandOkResponse`, use it instead of a generic `BsonDocument`.
-- Naming of callbacks.
-- Code formatting.
- remove unused parameters from `OperationHelper.canRetryWrite`/`canRetryRead`
- remove the broken `shouldAttemptToRetryWriteAndAddRetryableLabel` method
- remove the unused `RetryState.attempts` method, update docs

JAVA-5956, JAVA-6117, JAVA-6113, JAVA-6119, JAVA-6141

Author: stIncMale

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

@@ -56,19 +56,19 @@ public final class RetryState {
     private Throwable previouslyChosenException;
 
     /**
-     * Creates a {@code RetryState} with a positive number of allowed retries. {@link Integer#MAX_VALUE} is a special value interpreted as
-     * being unlimited.
+     * Creates a {@code RetryState} with a positive number of allowed retry attempts.
+     * {@link Integer#MAX_VALUE} is a special value interpreted as being unlimited.
      * <p>
-     * If a timeout is not specified in the {@link TimeoutContext#hasTimeoutMS()}, the specified {@code retries} param acts as a fallback
+     * If a timeout is not specified in the {@link TimeoutContext#hasTimeoutMS()}, the specified {@code retries} argument acts as a fallback
      * bound. Otherwise, retries are unbounded until the timeout is reached.
      * <p>
      * It is possible to provide an additional {@code retryPredicate} in the {@link #doAdvanceOrThrow} method,
-     * which can be used to stop retrying based on a custom condition additionally to {@code retires} and {@link TimeoutContext}.
+     * which can be used to stop retrying based on a custom condition additionally to {@code retries} and {@link TimeoutContext}.
      * </p>
      *
-     * @param retries A positive number of allowed retries. {@link Integer#MAX_VALUE} is a special value interpreted as being unlimited.
-     * @param retryUntilTimeoutThrowsException If {@code true}, then if a {@link MongoOperationTimeoutException} is throws then retrying stops.
-     * @see #attempts()
+     * @param retries A positive number of allowed retry attempts.
+     * {@link Integer#MAX_VALUE} is a special value interpreted as being unlimited.
+     * @param retryUntilTimeoutThrowsException If {@code true}, then if a {@link MongoOperationTimeoutException} is thrown then retrying stops.
      */
     public static RetryState withRetryableState(final int retries, final boolean retryUntilTimeoutThrowsException) {
         assertTrue(retries > 0);
@@ -80,24 +80,22 @@ public static RetryState withNonRetryableState() {
     }
 
     /**
-     * Creates a {@link RetryState} that does not limit the number of retries.
+     * Creates a {@link RetryState} that does not limit the number of attempts.
      * The number of attempts is limited iff {@link TimeoutContext#hasTimeoutMS()} is true and timeout has expired.
      * <p>
      * It is possible to provide an additional {@code retryPredicate} in the {@link #doAdvanceOrThrow} method,
-     * which can be used to stop retrying based on a custom condition additionally to {@code retires} and {@link TimeoutContext}.
+     * which can be used to stop retrying based on a custom condition additionally to {@link TimeoutContext}.
      * </p>
      *
      * @param timeoutContext A timeout context that will be used to determine if the operation has timed out.
-     * @see #attempts()
      */
     public RetryState(final TimeoutContext timeoutContext) {
         this(INFINITE_ATTEMPTS, timeoutContext.hasTimeoutMS());
     }
 
     /**
-     * @param retries A non-negative number of allowed retries. {@link Integer#MAX_VALUE} is a special value interpreted as being unlimited.
-     * @param retryUntilTimeoutThrowsException
-     * @see #attempts()
+     * @param retries A non-negative number of allowed retry attempts.
+     * {@link Integer#MAX_VALUE} is a special value interpreted as being unlimited.
      */
     private RetryState(final int retries, final boolean retryUntilTimeoutThrowsException) {
         assertTrue(retries >= 0);
@@ -108,7 +106,7 @@ private RetryState(final int retries, final boolean retryUntilTimeoutThrowsExcep
 
     /**
      * Advances this {@link RetryState} such that it represents the state of a new attempt.
-     * If there is at least one more {@linkplain #attempts() attempt} left, it is consumed by this method.
+     * If there is at least one more attempt left, it is consumed by this method.
      * Must not be called before the {@linkplain #isFirstAttempt() first attempt}, must be called before each subsequent attempt.
      * <p>
      * This method is intended to be used by code that generally does not handle {@link Error}s explicitly,
@@ -147,7 +145,7 @@ private RetryState(final int retries, final boolean retryUntilTimeoutThrowsExcep
      *     <li>the {@code retryPredicate} is {@code false}.</li>
      * </ul>
      * The exception thrown represents the failed result of the associated retryable activity,
-     * i.e., the caller must not do any more attempts.
+     * i.e., the caller must not make any more attempts.
      * @see #advanceOrThrow(Throwable, BinaryOperator, BiPredicate)
      */
     void advanceOrThrow(final RuntimeException attemptException, final BinaryOperator<Throwable> onAttemptFailureOperator,
@@ -354,58 +352,43 @@ public void markAsLastAttempt() {
     }
 
     /**
-     * Returns {@code true} iff the current attempt is the first one, i.e., no retries have been made.
+     * Returns {@code true} iff the current attempt is the first one, i.e., no retry attempts have been made.
      *
-     * @see #attempts()
+     * @see #attempt()
      */
     public boolean isFirstAttempt() {
         return loopState.isFirstIteration();
     }
 
     /**
-     * Returns {@code true} iff the current attempt is known to be the last one, i.e., it is known that no more retries will be made.
+     * Returns {@code true} iff the current attempt is known to be the last one, i.e., it is known that no more attempts will be made.
      * An attempt is known to be the last one iff any of the following applies:
      * <ul>
      *   <li>{@link #breakAndThrowIfRetryAnd(Supplier)} / {@link #breakAndCompleteIfRetryAnd(Supplier, SingleResultCallback)} / {@link #markAsLastAttempt()} was called.</li>
      *   <li>A timeout is set and has been reached.</li>
-     *   <li>No timeout is set, and the number of {@linkplain #attempts() attempts} is limited, and the current attempt is the last one.</li>
+     *   <li>No timeout is set, and the number of attempts is limited, and the current attempt is the last one.</li>
      * </ul>
      *
-     * @see #attempts()
+     * @see #attempt()
      */
     public boolean isLastAttempt() {
-        if (loopState.isLastIteration()){
+        if (loopState.isLastIteration()) {
             return true;
         }
         if (retryUntilTimeoutThrowsException) {
-           return false;
+            return false;
         }
         return attempt() == attempts - 1;
     }
 
     /**
      * A 0-based attempt number.
      *
-     * @see #attempts()
-     */
-    public int attempt() {
-        return loopState.iteration();
-    }
-
-    /**
-     * Returns a positive maximum number of attempts:
-     * <ul>
-     *     <li>0 if the number of retries is {@linkplain #RetryState(TimeoutContext) unlimited};</li>
-     *     <li>1 if no retries are allowed;</li>
-     *     <li>{@link #RetryState(int, boolean) retries} + 1 otherwise.</li>
-     * </ul>
-     *
-     * @see #attempt()
      * @see #isFirstAttempt()
      * @see #isLastAttempt()
      */
-    public int attempts() {
-        return attempts == INFINITE_ATTEMPTS ? 0 : attempts;
+    public int attempt() {
+        return loopState.iteration();
     }
 
     /**

driver-core/src/main/com/mongodb/internal/operation/AsyncOperationHelper.java

@@ -53,8 +53,8 @@
 import static com.mongodb.internal.operation.CommandOperationHelper.CommandCreator;
 import static com.mongodb.internal.operation.CommandOperationHelper.addRetryableWriteErrorLabel;
 import static com.mongodb.internal.operation.CommandOperationHelper.initialRetryState;
-import static com.mongodb.internal.operation.CommandOperationHelper.isRetryWritesEnabled;
-import static com.mongodb.internal.operation.CommandOperationHelper.logRetryExecute;
+import static com.mongodb.internal.operation.CommandOperationHelper.isRetryableWriteCommand;
+import static com.mongodb.internal.operation.CommandOperationHelper.logRetryCommand;
 import static com.mongodb.internal.operation.CommandOperationHelper.onRetryableReadAttemptFailure;
 import static com.mongodb.internal.operation.CommandOperationHelper.onRetryableWriteAttemptFailure;
 import static com.mongodb.internal.operation.CommandOperationHelper.transformWriteException;
@@ -199,9 +199,7 @@ static <D, T> void executeRetryableReadAsync(
                         withAsyncSourceAndConnection(sourceAsyncFunction, false, operationContext, funcCallback,
                                 (source, connection, operationContextWithMinRtt, releasingCallback) -> {
                                     if (retryState.breakAndCompleteIfRetryAnd(
-                                            () -> !OperationHelper.canRetryRead(source.getServerDescription(),
-                                                    operationContextWithMinRtt),
-                                            releasingCallback)) {
+                                            () -> !OperationHelper.canRetryRead(operationContextWithMinRtt), releasingCallback)) {
                                         return;
                                     }
                                     createReadCommandAndExecuteAsync(retryState, operationContextWithMinRtt, source, database,
@@ -278,8 +276,7 @@ static <T, R> void executeRetryableWriteAsync(
                                 ? releasingCallback
                                 : addingRetryableLabelCallback(releasingCallback, maxWireVersion);
                         if (retryState.breakAndCompleteIfRetryAnd(() ->
-                                        !OperationHelper.canRetryWrite(connection.getDescription(), operationContextWithMinRtt.getSessionContext()),
-                                addingRetryableLabelCallback)) {
+                                        !OperationHelper.canRetryWrite(connection.getDescription()), addingRetryableLabelCallback)) {
                             return;
                         }
                         BsonDocument command;
@@ -292,9 +289,9 @@ static <T, R> void executeRetryableWriteAsync(
                                             operationContextWithMinRtt,
                                             source.getServerDescription(),
                                             connection.getDescription()));
-                            // attach `maxWireVersion`, `retryableCommandFlag` ASAP because they are used to check whether we should retry
+                            // attach `maxWireVersion`, `retryableWriteCommandFlag` ASAP because they are used to check whether we should retry
                             retryState.attach(AttachmentKeys.maxWireVersion(), maxWireVersion, true)
-                                    .attach(AttachmentKeys.retryableCommandFlag(), isRetryWritesEnabled(command), true)
+                                    .attach(AttachmentKeys.retryableWriteCommandFlag(), isRetryableWriteCommand(command), true)
                                     .attach(AttachmentKeys.commandDescriptionSupplier(), command::getFirstKey, false)
                                     .attach(AttachmentKeys.command(), command, false);
                         } catch (Throwable t) {
@@ -335,8 +332,8 @@ static <D, T> void createReadCommandAndExecuteAsync(
     static <R> AsyncCallbackSupplier<R> decorateReadWithRetriesAsync(final RetryState retryState, final OperationContext operationContext,
             final AsyncCallbackSupplier<R> asyncReadFunction) {
         return new RetryingAsyncCallbackSupplier<>(retryState, onRetryableReadAttemptFailure(operationContext),
-                CommandOperationHelper::shouldAttemptToRetryRead, callback -> {
-            logRetryExecute(retryState, operationContext);
+                CommandOperationHelper::loggingShouldAttemptToRetryRead, callback -> {
+            logRetryCommand(retryState, operationContext);
             asyncReadFunction.get(callback);
         });
     }
@@ -345,7 +342,7 @@ static <R> AsyncCallbackSupplier<R> decorateWriteWithRetriesAsync(final RetrySta
             final AsyncCallbackSupplier<R> asyncWriteFunction) {
         return new RetryingAsyncCallbackSupplier<>(retryState, onRetryableWriteAttemptFailure(operationContext),
                 CommandOperationHelper::loggingShouldAttemptToRetryWriteAndAddRetryableLabel, callback -> {
-            logRetryExecute(retryState, operationContext);
+            logRetryCommand(retryState, operationContext);
             asyncWriteFunction.get(callback);
         });
     }