Add Kotlin Bulk Write API (#1591)

- Add sync and coroutine Bulk Write API.
- Enable unified spec tests.

JAVA-5532

Author: vbabanin

config/detekt/detekt.yml

@@ -159,7 +159,7 @@ complexity:
     active: true
     excludes: ['**/test/**']
     thresholdInFiles: 25
-    thresholdInClasses: 25
+    thresholdInClasses: 27
     thresholdInInterfaces: 25
     thresholdInObjects: 25
     thresholdInEnums: 25

driver-kotlin-coroutine/src/integration/kotlin/com/mongodb/kotlin/client/coroutine/syncadapter/SyncMongoCluster.kt

@@ -114,39 +114,25 @@ internal open class SyncMongoCluster(open val wrapped: MongoCluster) : JMongoClu
     ): ChangeStreamIterable<T> =
         SyncChangeStreamIterable(wrapped.watch(clientSession.unwrapped(), pipeline, resultClass))
 
-    override fun bulkWrite(models: MutableList<out ClientNamespacedWriteModel>): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
+    override fun bulkWrite(models: MutableList<out ClientNamespacedWriteModel>): ClientBulkWriteResult = runBlocking {
+        wrapped.bulkWrite(models)
     }
 
     override fun bulkWrite(
         models: MutableList<out ClientNamespacedWriteModel>,
         options: ClientBulkWriteOptions
-    ): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    ): ClientBulkWriteResult = runBlocking { wrapped.bulkWrite(models, options) }
 
     override fun bulkWrite(
         clientSession: ClientSession,
         models: MutableList<out ClientNamespacedWriteModel>
-    ): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    ): ClientBulkWriteResult = runBlocking { wrapped.bulkWrite(clientSession.unwrapped(), models) }
 
     override fun bulkWrite(
         clientSession: ClientSession,
         models: MutableList<out ClientNamespacedWriteModel>,
         options: ClientBulkWriteOptions
-    ): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    ): ClientBulkWriteResult = runBlocking { wrapped.bulkWrite(clientSession.unwrapped(), models, options) }
 
     private fun ClientSession.unwrapped() = (this as SyncClientSession).wrapped
 }

driver-kotlin-coroutine/src/main/kotlin/com/mongodb/kotlin/client/coroutine/MongoCluster.kt

@@ -15,12 +15,20 @@
  */
 package com.mongodb.kotlin.client.coroutine
 
+import com.mongodb.ClientBulkWriteException
 import com.mongodb.ClientSessionOptions
+import com.mongodb.MongoClientSettings
+import com.mongodb.MongoException
 import com.mongodb.ReadConcern
 import com.mongodb.ReadPreference
 import com.mongodb.WriteConcern
 import com.mongodb.annotations.Alpha
 import com.mongodb.annotations.Reason
+import com.mongodb.client.model.bulk.ClientBulkWriteOptions
+import com.mongodb.client.model.bulk.ClientBulkWriteResult
+import com.mongodb.client.model.bulk.ClientNamespacedDeleteManyModel
+import com.mongodb.client.model.bulk.ClientNamespacedUpdateManyModel
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel
 import com.mongodb.reactivestreams.client.MongoCluster as JMongoCluster
 import java.util.concurrent.TimeUnit
 import kotlinx.coroutines.flow.Flow
@@ -307,4 +315,111 @@ public open class MongoCluster protected constructor(private val wrapped: JMongo
         clientSession: ClientSession,
         pipeline: List<Bson> = emptyList()
     ): ChangeStreamFlow<T> = watch(clientSession, pipeline, T::class.java)
+
+    /**
+     * Executes a client-level bulk write operation. This method is functionally equivalent to
+     * [bulkWrite(models, options)][bulkWrite] with the
+     * [default options][ClientBulkWriteOptions.clientBulkWriteOptions].
+     *
+     * This operation supports [retryable writes][MongoClientSettings.getRetryWrites]. Depending on the number of
+     * `models`, encoded size of `models`, and the size limits in effect, executing this operation may require multiple
+     * `bulkWrite` commands. The eligibility for retries is determined per each `bulkWrite` command:
+     * [ClientNamespacedUpdateManyModel], [ClientNamespacedDeleteManyModel] in a command render it non-retryable.
+     *
+     * This operation is not supported by MongoDB Atlas Serverless instances.
+     *
+     * @param models The [individual write operations][ClientNamespacedWriteModel].
+     * @return The [ClientBulkWriteResult] if the operation is successful.
+     * @throws ClientBulkWriteException If and only if the operation is unsuccessful or partially unsuccessful, and
+     *   there is at least one of the following pieces of information to report:
+     *   [ClientBulkWriteException.getWriteConcernErrors], [ClientBulkWriteException.getWriteErrors],
+     *   [ClientBulkWriteException.getPartialResult].
+     * @throws MongoException Only if the operation is unsuccessful.
+     * @see [BulkWrite command](https://www.mongodb.com/docs/manual/reference/command/bulkWrite/)
+     * @since 5.3
+     */
+    public suspend fun bulkWrite(models: List<ClientNamespacedWriteModel>): ClientBulkWriteResult =
+        wrapped.bulkWrite(models).awaitSingle()
+
+    /**
+     * Executes a client-level bulk write operation.
+     *
+     * This operation supports [retryable writes][MongoClientSettings.getRetryWrites]. Depending on the number of
+     * `models`, encoded size of `models`, and the size limits in effect, executing this operation may require multiple
+     * `bulkWrite` commands. The eligibility for retries is determined per each `bulkWrite` command:
+     * [ClientNamespacedUpdateManyModel], [ClientNamespacedDeleteManyModel] in a command render it non-retryable.
+     *
+     * This operation is not supported by MongoDB Atlas Serverless instances.
+     *
+     * @param models The [individual write operations][ClientNamespacedWriteModel].
+     * @param options The [options][ClientBulkWriteOptions].
+     * @return The [ClientBulkWriteResult] if the operation is successful.
+     * @throws ClientBulkWriteException If and only if the operation is unsuccessful or partially unsuccessful, and
+     *   there is at least one of the following pieces of information to report:
+     *   [ClientBulkWriteException.getWriteConcernErrors], [ClientBulkWriteException.getWriteErrors],
+     *   [ClientBulkWriteException.getPartialResult].
+     * @throws MongoException Only if the operation is unsuccessful.
+     * @see [BulkWrite command](https://www.mongodb.com/docs/manual/reference/command/bulkWrite/)
+     * @since 5.3
+     */
+    public suspend fun bulkWrite(
+        models: List<ClientNamespacedWriteModel>,
+        options: ClientBulkWriteOptions
+    ): ClientBulkWriteResult = wrapped.bulkWrite(models, options).awaitSingle()
+
+    /**
+     * Executes a client-level bulk write operation. This method is functionally equivalent to
+     * [bulkWrite(clientSession, models, options)][bulkWrite] with the
+     * [default options][ClientBulkWriteOptions.clientBulkWriteOptions].
+     *
+     * This operation supports [retryable writes][MongoClientSettings.getRetryWrites]. Depending on the number of
+     * `models`, encoded size of `models`, and the size limits in effect, executing this operation may require multiple
+     * `bulkWrite` commands. The eligibility for retries is determined per each `bulkWrite` command:
+     * [ClientNamespacedUpdateManyModel], [ClientNamespacedDeleteManyModel] in a command render it non-retryable.
+     *
+     * This operation is not supported by MongoDB Atlas Serverless instances.
+     *
+     * @param clientSession The [client session][ClientSession] with which to associate this operation.
+     * @param models The [individual write operations][ClientNamespacedWriteModel].
+     * @return The [ClientBulkWriteResult] if the operation is successful.
+     * @throws ClientBulkWriteException If and only if the operation is unsuccessful or partially unsuccessful, and
+     *   there is at least one of the following pieces of information to report:
+     *   [ClientBulkWriteException.getWriteConcernErrors], [ClientBulkWriteException.getWriteErrors],
+     *   [ClientBulkWriteException.getPartialResult].
+     * @throws MongoException Only if the operation is unsuccessful.
+     * @see [BulkWrite command](https://www.mongodb.com/docs/manual/reference/command/bulkWrite/)
+     * @since 5.3
+     */
+    public suspend fun bulkWrite(
+        clientSession: ClientSession,
+        models: List<ClientNamespacedWriteModel>
+    ): ClientBulkWriteResult = wrapped.bulkWrite(clientSession.wrapped, models).awaitSingle()
+
+    /**
+     * Executes a client-level bulk write operation.
+     *
+     * This operation supports [retryable writes][MongoClientSettings.getRetryWrites]. Depending on the number of
+     * `models`, encoded size of `models`, and the size limits in effect, executing this operation may require multiple
+     * `bulkWrite` commands. The eligibility for retries is determined per each `bulkWrite` command:
+     * [ClientNamespacedUpdateManyModel], [ClientNamespacedDeleteManyModel] in a command render it non-retryable.
+     *
+     * This operation is not supported by MongoDB Atlas Serverless instances.
+     *
+     * @param clientSession The [client session][ClientSession] with which to associate this operation.
+     * @param models The [individual write operations][ClientNamespacedWriteModel].
+     * @param options The [options][ClientBulkWriteOptions].
+     * @return The [ClientBulkWriteResult] if the operation is successful.
+     * @throws ClientBulkWriteException If and only if the operation is unsuccessful or partially unsuccessful, and
+     *   there is at least one of the following pieces of information to report:
+     *   [ClientBulkWriteException.getWriteConcernErrors], [ClientBulkWriteException.getWriteErrors],
+     *   [ClientBulkWriteException.getPartialResult].
+     * @throws MongoException Only if the operation is unsuccessful.
+     * @see [BulkWrite command](https://www.mongodb.com/docs/manual/reference/command/bulkWrite/)
+     * @since 5.3
+     */
+    public suspend fun bulkWrite(
+        clientSession: ClientSession,
+        models: List<ClientNamespacedWriteModel>,
+        options: ClientBulkWriteOptions
+    ): ClientBulkWriteResult = wrapped.bulkWrite(clientSession.wrapped, models, options).awaitSingle()
 }

driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/MongoClientTest.kt

@@ -16,6 +16,9 @@
 package com.mongodb.kotlin.client.coroutine
 
 import com.mongodb.ClientSessionOptions
+import com.mongodb.MongoNamespace
+import com.mongodb.client.model.bulk.ClientBulkWriteOptions
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel
 import com.mongodb.reactivestreams.client.MongoClient as JMongoClient
 import kotlin.reflect.full.declaredFunctions
 import kotlin.test.assertEquals
@@ -166,4 +169,29 @@ class MongoClientTest {
         verify(wrapped, times(2)).watch(clientSession.wrapped, pipeline, BsonDocument::class.java)
         verifyNoMoreInteractions(wrapped)
     }
+
+    @Test
+    fun shouldCallTheUnderlyingBulkWrite() {
+        val mongoClient = MongoClient(wrapped)
+        val requests = listOf(ClientNamespacedWriteModel.insertOne(MongoNamespace("test.test"), Document()))
+        val options = ClientBulkWriteOptions.clientBulkWriteOptions().bypassDocumentValidation(true)
+
+        whenever(wrapped.bulkWrite(requests)).doReturn(Mono.fromCallable { mock() })
+        whenever(wrapped.bulkWrite(requests, options)).doReturn(Mono.fromCallable { mock() })
+        whenever(wrapped.bulkWrite(clientSession.wrapped, requests)).doReturn(Mono.fromCallable { mock() })
+        whenever(wrapped.bulkWrite(clientSession.wrapped, requests, options)).doReturn(Mono.fromCallable { mock() })
+
+        runBlocking {
+            mongoClient.bulkWrite(requests)
+            mongoClient.bulkWrite(requests, options)
+            mongoClient.bulkWrite(clientSession, requests)
+            mongoClient.bulkWrite(clientSession, requests, options)
+        }
+
+        verify(wrapped).bulkWrite(requests)
+        verify(wrapped).bulkWrite(requests, options)
+        verify(wrapped).bulkWrite(clientSession.wrapped, requests)
+        verify(wrapped).bulkWrite(clientSession.wrapped, requests, options)
+        verifyNoMoreInteractions(wrapped)
+    }
 }

driver-kotlin-sync/src/integration/kotlin/com/mongodb/kotlin/client/syncadapter/SyncMongoCluster.kt

@@ -113,39 +113,24 @@ internal open class SyncMongoCluster(open val wrapped: MongoCluster) : JMongoClu
     ): ChangeStreamIterable<T> =
         SyncChangeStreamIterable(wrapped.watch(clientSession.unwrapped(), pipeline, resultClass))
 
-    override fun bulkWrite(models: MutableList<out ClientNamespacedWriteModel>): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    override fun bulkWrite(models: MutableList<out ClientNamespacedWriteModel>): ClientBulkWriteResult =
+        wrapped.bulkWrite(models)
 
     override fun bulkWrite(
         models: MutableList<out ClientNamespacedWriteModel>,
         options: ClientBulkWriteOptions
-    ): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    ): ClientBulkWriteResult = wrapped.bulkWrite(models, options)
 
     override fun bulkWrite(
         clientSession: ClientSession,
         models: MutableList<out ClientNamespacedWriteModel>
-    ): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    ): ClientBulkWriteResult = wrapped.bulkWrite(clientSession.unwrapped(), models)
 
     override fun bulkWrite(
         clientSession: ClientSession,
         models: MutableList<out ClientNamespacedWriteModel>,
         options: ClientBulkWriteOptions
-    ): ClientBulkWriteResult {
-        org.junit.jupiter.api.Assumptions.assumeTrue(
-            java.lang.Boolean.parseBoolean(toString()), "BULK-TODO Kotlin implement")
-        TODO("BULK-TODO Kotlin implement")
-    }
+    ): ClientBulkWriteResult = wrapped.bulkWrite(clientSession.unwrapped(), models, options)
 
     private fun ClientSession.unwrapped() = (this as SyncClientSession).wrapped
 }