Improved Bulk Write API

bson/src/main/org/bson/AbstractBsonWriter.java

@@ -748,6 +748,15 @@ protected void throwInvalidState(final String methodName, final State... validSt
                                                        methodName, validStatesString, state));
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * The {@link #flush()} method of {@link AbstractBsonWriter} does nothing.</p>
+     */
+    @Override
+    public void flush() {
+    }
+
     @Override
     public void close() {
         closed = true;

bson/src/main/org/bson/BSONCallbackAdapter.java

@@ -39,11 +39,6 @@ protected BSONCallbackAdapter(final BsonWriterSettings settings, final BSONCallb
         this.bsonCallback = bsonCallback;
     }
 
-    @Override
-    public void flush() {
-        //Looks like should be no-op?
-    }
-
     @Override
     public void doWriteStartDocument() {
         BsonContextType contextType = getState() == State.SCOPE_DOCUMENT

bson/src/main/org/bson/BsonBinaryWriter.java

@@ -108,10 +108,6 @@ public BsonBinaryWriterSettings getBinaryWriterSettings() {
         return binaryWriterSettings;
     }
 
-    @Override
-    public void flush() {
-    }
-
     @Override
     protected Context getContext() {
         return (Context) super.getContext();

bson/src/main/org/bson/BsonDocumentWriter.java

@@ -194,10 +194,6 @@ public void doWriteUndefined() {
         write(new BsonUndefined());
     }
 
-    @Override
-    public void flush() {
-    }
-
     @Override
     protected Context getContext() {
         return (Context) super.getContext();

bson/src/main/org/bson/io/OutputBuffer.java

@@ -41,6 +41,16 @@ public void write(final byte[] b) {
     public void close() {
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * The {@link #flush()} method of {@link OutputBuffer} does nothing.</p>
+     */
+    @Override
+    public void flush() throws IOException {
+        super.flush();
+    }
+
     @Override
     public void write(final byte[] bytes, final int offset, final int length) {
         writeBytes(bytes, offset, length);

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

@@ -0,0 +1,154 @@
+/*
+ * 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;
+
+import com.mongodb.bulk.WriteConcernError;
+import com.mongodb.client.model.bulk.ClientBulkWriteResult;
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
+import com.mongodb.lang.Nullable;
+
+import java.util.List;
+import java.util.Map;
+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;
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Optional.ofNullable;
+
+/**
+ * The result of an unsuccessful or partially unsuccessful client-level bulk write operation.
+ * Note that the {@linkplain #getCode() code} and {@linkplain #getErrorLabels() labels} from this exception are not useful.
+ * An application should use those from the {@linkplain #getError() top-level error}.
+ *
+ * @see ClientBulkWriteResult
+ * @since 5.3
+ * @serial exclude
+ */
+public final class ClientBulkWriteException extends MongoServerException {
+    private static final long serialVersionUID = 1;
+
+    @Nullable
+    private final MongoException error;
+    private final List<WriteConcernError> writeConcernErrors;
+    private final Map<Integer, WriteError> writeErrors;
+    @Nullable
+    private final ClientBulkWriteResult partialResult;
+
+    /**
+     * Constructs a new instance.
+     *
+     * @param error The {@linkplain #getError() top-level error}.
+     * @param writeConcernErrors The {@linkplain #getWriteConcernErrors() write concern errors}.
+     * @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,
+                        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())
+                        || partialResult != null);
+        this.error = error;
+        this.writeConcernErrors = writeConcernErrors == null ? emptyList() : unmodifiableList(writeConcernErrors);
+        this.writeErrors = writeErrors == null ? emptyMap() : unmodifiableMap(writeErrors);
+        this.partialResult = partialResult;
+    }
+
+    private static String message(
+            @Nullable final MongoException error,
+            @Nullable final List<WriteConcernError> writeConcernErrors,
+            @Nullable final Map<Integer, WriteError> writeErrors,
+            @Nullable final ClientBulkWriteResult partialResult,
+            final ServerAddress serverAddress) {
+        return "Client-level bulk write operation error on server " + serverAddress + "."
+                + (error == null ? "" : " Top-level error: " + error + ".")
+                + (writeErrors == null || writeErrors.isEmpty() ? "" : " Write errors: " + writeErrors + ".")
+                + (writeConcernErrors == null || writeConcernErrors.isEmpty() ? "" : " Write concern errors: " + writeConcernErrors + ".")
+                + (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}.
+     *
+     * @return The top-level error. {@linkplain Optional#isPresent() Present} only if a top-level error occurred.
+     */
+    public Optional<MongoException> getError() {
+        return ofNullable(error);
+    }
+
+    /**
+     * The {@link WriteConcernError}s that occurred while executing the client-level bulk write operation.
+     * <p>
+     * There are no guarantees on mutability of the {@link List} returned.</p>
+     *
+     * @return The {@link WriteConcernError}s.
+     */
+    public List<WriteConcernError> getWriteConcernErrors() {
+        return writeConcernErrors;
+    }
+
+    /**
+     * The indexed {@link WriteError}s.
+     * The {@linkplain Map#keySet() keys} are the indexes of the corresponding {@link ClientNamespacedWriteModel}s
+     * in the corresponding client-level bulk write operation.
+     * <p>
+     * There are no guarantees on mutability or iteration order of the {@link Map} returned.</p>
+     *
+     * @return The indexed {@link WriteError}s.
+     * @see ClientBulkWriteResult.VerboseResults#getInsertResults()
+     * @see ClientBulkWriteResult.VerboseResults#getUpdateResults()
+     * @see ClientBulkWriteResult.VerboseResults#getDeleteResults()
+     */
+    public Map<Integer, WriteError> getWriteErrors() {
+        return writeErrors;
+    }
+
+    /**
+     * The result of the part of a client-level bulk write operation that is known to be successful.
+     *
+     * @return The successful partial result. {@linkplain Optional#isPresent() Present} only if the client received a response indicating success
+     * of at least one {@linkplain ClientNamespacedWriteModel individual write operation}.
+     */
+    public Optional<ClientBulkWriteResult> getPartialResult() {
+        return ofNullable(partialResult);
+    }
+}

driver-core/src/main/com/mongodb/client/model/bulk/ClientBulkWriteOptions.java

@@ -0,0 +1,90 @@
+/*
+ * 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.client.model.bulk;
+
+import com.mongodb.annotations.Sealed;
+import com.mongodb.client.model.Filters;
+import com.mongodb.internal.client.model.bulk.ConcreteClientBulkWriteOptions;
+import com.mongodb.lang.Nullable;
+import org.bson.BsonValue;
+import org.bson.conversions.Bson;
+
+/**
+ * The options to apply when executing a client-level bulk write operation.
+ *
+ * @since 5.3
+ */
+@Sealed
+public interface ClientBulkWriteOptions {
+    /**
+     * Creates the default options.
+     *
+     * @return The default options.
+     */
+    static ClientBulkWriteOptions clientBulkWriteOptions() {
+        return new ConcreteClientBulkWriteOptions();
+    }
+
+    /**
+     * Enables or disables ordered execution of {@linkplain ClientNamespacedWriteModel individual write operations}.
+     * In an ordered execution a failure of an individual operation prevents the rest of them
+     * from being executed.
+     * In an unordered execution failures of individual operations do not prevent the rest of them
+     * from being executed.
+     *
+     * @param ordered The ordered flag. If {@code null}, the client defaults to {@code true}.
+     * @return {@code this}.
+     */
+    ClientBulkWriteOptions ordered(@Nullable Boolean ordered);
+
+    /**
+     * Disables or enables checking against document validation rules, a.k.a., schema validation.
+     *
+     * @param bypassDocumentValidation The flag specifying whether to bypass the document validation rules.
+     * {@code null} represents the server default.
+     * @return {@code this}.
+     */
+    ClientBulkWriteOptions bypassDocumentValidation(@Nullable Boolean bypassDocumentValidation);
+
+    /**
+     * Sets variables that can be referenced from {@linkplain ClientNamespacedWriteModel individual write operations}
+     * with the {@code "$$"} syntax, which in turn requires using {@link Filters#expr(Object)} when specifying filters.
+     * Values must be constants or expressions that do not reference fields.
+     *
+     * @param let The variables. {@code null} represents the server default.
+     * @return {@code this}.
+     * @mongodb.driver.manual reference/aggregation-variables/ Variables in Aggregation Expressions
+     */
+    ClientBulkWriteOptions let(@Nullable Bson let);
+
+    /**
+     * Sets the comment to attach to the {@code bulkWrite} administration command.
+     *
+     * @param comment The comment. {@code null} represents the server default.
+     * @return {@code this}.
+     */
+    ClientBulkWriteOptions comment(@Nullable BsonValue comment);
+
+    /**
+     * Enables or disables requesting {@linkplain ClientBulkWriteResult#getVerboseResults() verbose results}.
+     *
+     * @param verboseResults The flag specifying whether to request verbose results.
+     * If {@code null}, the client defaults to {@code false}.
+     * This value corresponds inversely to the {@code errorsOnly} field of the {@code bulkWrite} administration command.
+     * @return {@code this}.
+     */
+    ClientBulkWriteOptions verboseResults(@Nullable Boolean verboseResults);
+}