Add the ClientEncryption.createEncryptedCollection helper method

config/spotbugs/exclude.xml

@@ -240,4 +240,18 @@
         <Method name="acquirePermitOrGetAvailableOpenedConnection"/>
         <Bug pattern="NS_NON_SHORT_CIRCUIT"/>
     </Match>
+
+    <!-- Can actually be null, but is not annotated as `@Nullable`. Annotating it as such causes warnings
+         in other places where `null` is not handled, see https://jira.mongodb.org/browse/JAVA-4861.
+         When the aforementioned ticket is done, it will be clear what to do with the warnings suppressed here. -->
+    <Match>
+        <Class name="com.mongodb.client.internal.ClientEncryptionImpl"/>
+        <Method name="createEncryptedCollection"/>
+        <Bug pattern="RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE"/>
+    </Match>
+    <Match>
+        <Class name="com.mongodb.reactivestreams.client.internal.vault.ClientEncryptionImpl"/>
+        <Method name="~.*createEncryptedCollection.*"/>
+        <Bug pattern="RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE"/>
+    </Match>
 </FindBugsFilter>

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

@@ -0,0 +1,67 @@
+/*
+ * 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.annotations.Beta;
+import org.bson.BsonDocument;
+
+import static com.mongodb.assertions.Assertions.assertNotNull;
+
+/**
+ * An exception thrown by methods that may automatically create data encryption keys
+ * where needed based on the {@code encryptedFields} configuration.
+ *
+ * @since 4.9
+ */
+@Beta(Beta.Reason.SERVER)
+public final class MongoUpdatedEncryptedFieldsException extends MongoClientException {
+    private static final long serialVersionUID = 1;
+
+    private final BsonDocument encryptedFields;
+
+    /**
+     * Not part of the public API.
+     *
+     * @param encryptedFields The (partially) updated {@code encryptedFields} document,
+     * which allows users to infer which data keys are known to be created before the exception happened
+     * (see {@link #getEncryptedFields()} for more details).
+     * Reporting this back to a user may be helpful because creation of a data key includes persisting it in the key vault.
+     * @param msg The message.
+     * @param cause The cause.
+     */
+    public MongoUpdatedEncryptedFieldsException(final BsonDocument encryptedFields, final String msg, final Throwable cause) {
+        super(msg, assertNotNull(cause));
+        this.encryptedFields = assertNotNull(encryptedFields);
+    }
+
+    /**
+     * The {@code encryptedFields} document that allows inferring which data keys are <strong>known to be created</strong>
+     * before {@code this} exception happened by comparing this document with the original {@code encryptedFields} configuration.
+     * Creation of a data key includes persisting it in the key vault.
+     * <p>
+     * Note that the returned {@code encryptedFields} document is not guaranteed to contain information about all the data keys that
+     * may be created, only about those that the driver is certain about. For example, if persisting a data key times out,
+     * the driver does not know whether it can be considered created or not, and does not include the information about the key in
+     * the {@code encryptedFields} document. You can analyze whether the {@linkplain #getCause() cause} is a definite or indefinite
+     * error, and rely on the returned {@code encryptedFields} to be containing information on all created keys
+     * only if the error is definite.</p>
+     *
+     * @return The updated {@code encryptedFields} document.
+     */
+    public BsonDocument getEncryptedFields() {
+        return encryptedFields;
+    }
+}

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

@@ -22,6 +22,7 @@
 
 import java.util.concurrent.TimeUnit;
 
+import static com.mongodb.assertions.Assertions.fail;
 import static com.mongodb.assertions.Assertions.notNull;
 
 /**
@@ -31,7 +32,7 @@
  * @mongodb.driver.manual core/timeseries-collections/ Time-series collections
  * @since 3.0
  */
-public class CreateCollectionOptions {
+public class CreateCollectionOptions implements Cloneable {
     private long maxDocuments;
     private boolean capped;
     private long sizeInBytes;
@@ -360,4 +361,19 @@ public String toString() {
                 + ", encryptedFields=" + encryptedFields
                 + '}';
     }
+
+    /**
+     * Creates a shallow copy of {@code this} {@link CreateCollectionOptions} by calling {@code super.clone()}.
+     *
+     * @return A shallow copy of {@code this} {@link CreateCollectionOptions}.
+     * @since 4.9
+     */
+    @Override
+    public CreateCollectionOptions clone() {
+        try {
+            return (CreateCollectionOptions) super.clone();
+        } catch (CloneNotSupportedException e) {
+            throw fail(e.toString());
+        }
+    }
 }

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

@@ -0,0 +1,85 @@
+/*
+ * 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;
+
+import com.mongodb.annotations.Beta;
+import com.mongodb.client.model.vault.DataKeyOptions;
+import com.mongodb.lang.Nullable;
+import org.bson.BsonDocument;
+
+import static com.mongodb.assertions.Assertions.notNull;
+
+/**
+ * Auxiliary parameters for creating an encrypted collection.
+ *
+ * @since 4.9
+ */
+@Beta(Beta.Reason.SERVER)
+public final class CreateEncryptedCollectionParams {
+    private final String kmsProvider;
+    @Nullable
+    private BsonDocument masterKey;
+
+    /**
+     * A constructor.
+     *
+     * @param kmsProvider The name of the KMS provider.
+     */
+    public CreateEncryptedCollectionParams(final String kmsProvider) {
+        this.kmsProvider = notNull("kmsProvider", kmsProvider);
+        masterKey = null;
+    }
+
+    /**
+     * The name of the KMS provider.
+     *
+     * @return The name of the KMS provider.
+     */
+    public String getKmsProvider() {
+        return kmsProvider;
+    }
+
+    /**
+     * Sets the {@linkplain DataKeyOptions#getMasterKey() master key} for creating a data key.
+     *
+     * @param masterKey The master key for creating a data key.
+     * @return {@code this}.
+     */
+    public CreateEncryptedCollectionParams masterKey(@Nullable final BsonDocument masterKey) {
+        this.masterKey = masterKey;
+        return this;
+    }
+
+    /**
+     * The {@linkplain DataKeyOptions#getMasterKey() master key} for creating a data key.
+     * The default is {@code null}.
+     *
+     * @return The master key for creating a data key.
+     */
+    @Nullable
+    public BsonDocument getMasterKey() {
+        return masterKey;
+    }
+
+    @Override
+    public String toString() {
+        return "CreateEncryptedCollectionParams{"
+                + ", kmsProvider=" + kmsProvider
+                + ", masterKey=" + masterKey
+                + '}';
+    }
+}

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OneShotPublisher.java

@@ -0,0 +1,69 @@
+/*
+ * 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.reactivestreams.client.internal;
+
+import com.mongodb.annotations.NotThreadSafe;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Supplier;
+
+/**
+ * Allows creating {@link Publisher}s that do not allow calling {@link Publisher#subscribe(Subscriber)} more than once.
+ *
+ * <p>This class is not part of the public API and may be removed or changed at any time</p>
+ */
+public final class OneShotPublisher {
+    public static <T> Publisher<T> from(final Publisher<T> publisher) {
+        return Flux.defer(new OneShotSupplier<>(publisher));
+    }
+
+    // A `Publisher` does not have to be thread-safe, therefore, `Publisher.subscribe` should not be called concurrently.
+    // Hence, `OneShotSupplier.get` is not called concurrently and does not have to be thread-safe.
+    @NotThreadSafe
+    private static final class OneShotSupplier<T> implements Supplier<Publisher<T>> {
+        private final Publisher<T> publisher;
+        private boolean used;
+
+        OneShotSupplier(final Publisher<T> publisher) {
+            this.publisher = publisher;
+        }
+
+        @Override
+        public Publisher<T> get() {
+            if (used) {
+                // we may also `throw` here, and `Flux.defer` will handle the exception and signal `onSubscribe` followed by `onError`
+                return Mono.error(new IllegalStateException(
+                        "This is a one-shot publisher, it does not support subscribing to it more than once."));
+            }
+            used = true;
+            return publisher;
+        }
+
+        @Override
+        public String toString() {
+            return "OneShotSupplier{"
+                    + "publisher=" + publisher
+                    + ", used=" + used
+                    + '}';
+        }
+    }
+
+    private OneShotPublisher() {
+    }
+}