Implement Java sync improved bulk write API and unified spec tests

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

@@ -25,6 +25,8 @@
 import java.util.Optional;
 
 import static com.mongodb.assertions.Assertions.isTrueArgument;
+import static com.mongodb.assertions.Assertions.notNull;
+import static com.mongodb.internal.operation.ClientBulkWriteOperation.Exceptions.serverAddressFromException;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.unmodifiableList;
@@ -58,14 +60,20 @@ public final class ClientBulkWriteException extends MongoServerException {
      * @param writeErrors The {@linkplain #getWriteErrors() write errors}.
      * @param partialResult The {@linkplain #getPartialResult() partial result}.
      * @param serverAddress The {@linkplain MongoServerException#getServerAddress() server address}.
+     * If {@code error} is a {@link MongoServerException} or a {@link MongoSocketException}, then {@code serverAddress}
+     * must be equal to the {@link ServerAddress} they bear.
      */
     public ClientBulkWriteException(
             @Nullable final MongoException error,
             @Nullable final List<WriteConcernError> writeConcernErrors,
             @Nullable final Map<Integer, WriteError> writeErrors,
             @Nullable final ClientBulkWriteResult partialResult,
             final ServerAddress serverAddress) {
-        super(message(error, writeConcernErrors, writeErrors, partialResult, serverAddress), serverAddress);
+        super(
+                message(
+                        error, writeConcernErrors, writeErrors, partialResult,
+                        notNull("serverAddress", serverAddress)),
+                validateServerAddress(error, serverAddress));
         isTrueArgument("At least one of `writeConcernErrors`, `writeErrors`, `partialResult` must be non-null or non-empty",
                 !(writeConcernErrors == null || writeConcernErrors.isEmpty())
                         || !(writeErrors == null || writeErrors.isEmpty())
@@ -89,6 +97,14 @@ private static String message(
                 + (partialResult == null ? "" : " Partial result: " + partialResult + ".");
     }
 
+    private static ServerAddress validateServerAddress(@Nullable final MongoException error, final ServerAddress serverAddress) {
+        serverAddressFromException(error).ifPresent(serverAddressFromError ->
+                isTrueArgument("`serverAddress` must be equal to that of the `error`", serverAddressFromError.equals(serverAddress)));
+        return error instanceof MongoServerException
+                ? ((MongoServerException) error).getServerAddress()
+                : serverAddress;
+    }
+
     /**
      * The top-level error. That is an error that is neither a {@linkplain #getWriteConcernErrors() write concern error},
      * nor is an {@linkplain #getWriteErrors() error of an individual write operation}.

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

@@ -322,7 +322,7 @@ static <R> AsyncCallbackSupplier<R> decorateReadWithRetriesAsync(final RetryStat
     static <R> AsyncCallbackSupplier<R> decorateWriteWithRetriesAsync(final RetryState retryState, final OperationContext operationContext,
             final AsyncCallbackSupplier<R> asyncWriteFunction) {
         return new RetryingAsyncCallbackSupplier<>(retryState, onRetryableWriteAttemptFailure(operationContext),
-                CommandOperationHelper::shouldAttemptToRetryWrite, callback -> {
+                CommandOperationHelper::loggingShouldAttemptToRetryWriteAndAddRetryableLabel, callback -> {
             logRetryExecute(retryState, operationContext);
             asyncWriteFunction.get(callback);
         });

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

@@ -64,7 +64,7 @@
 import static com.mongodb.internal.bulk.WriteRequest.Type.REPLACE;
 import static com.mongodb.internal.bulk.WriteRequest.Type.UPDATE;
 import static com.mongodb.internal.operation.DocumentHelper.putIfNotNull;
-import static com.mongodb.internal.operation.MixedBulkWriteOperation.commandWriteConcern;
+import static com.mongodb.internal.operation.CommandOperationHelper.commandWriteConcern;
 import static com.mongodb.internal.operation.OperationHelper.LOGGER;
 import static com.mongodb.internal.operation.OperationHelper.isRetryableWrite;
 import static com.mongodb.internal.operation.WriteConcernHelper.createWriteConcernError;
@@ -111,7 +111,7 @@ static BulkWriteBatch createBulkWriteBatch(final MongoNamespace namespace,
         }
         if (canRetryWrites && !writeRequestsAreRetryable) {
             canRetryWrites = false;
-            LOGGER.debug("retryWrites set but one or more writeRequests do not support retryable writes");
+            logWriteModelDoesNotSupportRetries();
         }
         return new BulkWriteBatch(namespace, connectionDescription, ordered, writeConcern, bypassDocumentValidation,
                 canRetryWrites, new BulkWriteBatchCombiner(connectionDescription.getServerAddress(), ordered, writeConcern),
@@ -385,4 +385,8 @@ private static boolean isRetryable(final WriteRequest writeRequest) {
         }
         return true;
     }
+
+    static void logWriteModelDoesNotSupportRetries() {
+        LOGGER.debug("retryWrites set but one or more writeRequests do not support retryable writes");
+    }
 }