diff --git a/config/detekt/detekt.yml b/config/detekt/detekt.yml
index 4c083b0bce5..4ac460b0738 100644
--- a/config/detekt/detekt.yml
+++ b/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
diff --git a/driver-kotlin-coroutine/src/integration/kotlin/com/mongodb/kotlin/client/coroutine/syncadapter/SyncMongoCluster.kt b/driver-kotlin-coroutine/src/integration/kotlin/com/mongodb/kotlin/client/coroutine/syncadapter/SyncMongoCluster.kt
index 2c377e41d41..01d0e27ff58 100644
--- a/driver-kotlin-coroutine/src/integration/kotlin/com/mongodb/kotlin/client/coroutine/syncadapter/SyncMongoCluster.kt
+++ b/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
 }
diff --git a/driver-kotlin-coroutine/src/main/kotlin/com/mongodb/kotlin/client/coroutine/MongoCluster.kt b/driver-kotlin-coroutine/src/main/kotlin/com/mongodb/kotlin/client/coroutine/MongoCluster.kt
index 88df39dd23d..65ec0aa7f45 100644
--- a/driver-kotlin-coroutine/src/main/kotlin/com/mongodb/kotlin/client/coroutine/MongoCluster.kt
+++ b/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()
 }
diff --git a/driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/MongoClientTest.kt b/driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/MongoClientTest.kt
index 9ac4805f6fa..fd66e4de31b 100644
--- a/driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/MongoClientTest.kt
+++ b/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)
+    }
 }
diff --git a/driver-kotlin-sync/src/integration/kotlin/com/mongodb/kotlin/client/syncadapter/SyncMongoCluster.kt b/driver-kotlin-sync/src/integration/kotlin/com/mongodb/kotlin/client/syncadapter/SyncMongoCluster.kt
index a4ad9bd1418..b86f2447a17 100644
--- a/driver-kotlin-sync/src/integration/kotlin/com/mongodb/kotlin/client/syncadapter/SyncMongoCluster.kt
+++ b/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
 }
diff --git a/driver-kotlin-sync/src/main/kotlin/com/mongodb/kotlin/client/MongoCluster.kt b/driver-kotlin-sync/src/main/kotlin/com/mongodb/kotlin/client/MongoCluster.kt
index f541aaf1a9f..90fb45d1dbd 100644
--- a/driver-kotlin-sync/src/main/kotlin/com/mongodb/kotlin/client/MongoCluster.kt
+++ b/driver-kotlin-sync/src/main/kotlin/com/mongodb/kotlin/client/MongoCluster.kt
@@ -15,13 +15,21 @@
  */
 package com.mongodb.kotlin.client
 
+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.MongoCluster as JMongoCluster
+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 java.util.concurrent.TimeUnit
 import org.bson.Document
 import org.bson.codecs.configuration.CodecRegistry
@@ -303,4 +311,110 @@ public open class MongoCluster protected constructor(private val wrapped: JMongo
         clientSession: ClientSession,
         pipeline: List<Bson> = emptyList()
     ): ChangeStreamIterable<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 fun bulkWrite(models: List<ClientNamespacedWriteModel>): ClientBulkWriteResult = wrapped.bulkWrite(models)
+
+    /**
+     * 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 fun bulkWrite(
+        models: List<ClientNamespacedWriteModel>,
+        options: ClientBulkWriteOptions
+    ): ClientBulkWriteResult = wrapped.bulkWrite(models, options)
+
+    /**
+     * 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 fun bulkWrite(
+        clientSession: ClientSession,
+        models: List<ClientNamespacedWriteModel>
+    ): ClientBulkWriteResult = wrapped.bulkWrite(clientSession.wrapped, models)
+
+    /**
+     * Executes a client-level bulk write operation.
+     *
+     * This operation supports [retryable writes][com.mongodb.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 fun bulkWrite(
+        clientSession: ClientSession,
+        models: List<ClientNamespacedWriteModel>,
+        options: ClientBulkWriteOptions
+    ): ClientBulkWriteResult = wrapped.bulkWrite(clientSession.wrapped, models, options)
 }
diff --git a/driver-kotlin-sync/src/test/kotlin/com/mongodb/kotlin/client/MongoClientTest.kt b/driver-kotlin-sync/src/test/kotlin/com/mongodb/kotlin/client/MongoClientTest.kt
index 0999e77080e..0aa0c582ff4 100644
--- a/driver-kotlin-sync/src/test/kotlin/com/mongodb/kotlin/client/MongoClientTest.kt
+++ b/driver-kotlin-sync/src/test/kotlin/com/mongodb/kotlin/client/MongoClientTest.kt
@@ -16,7 +16,10 @@
 package com.mongodb.kotlin.client
 
 import com.mongodb.ClientSessionOptions
+import com.mongodb.MongoNamespace
 import com.mongodb.client.MongoClient as JMongoClient
+import com.mongodb.client.model.bulk.ClientBulkWriteOptions
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel
 import kotlin.reflect.full.declaredFunctions
 import kotlin.reflect.full.declaredMemberProperties
 import kotlin.test.assertEquals
@@ -168,4 +171,27 @@ 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(mock())
+        whenever(wrapped.bulkWrite(requests, options)).doReturn(mock())
+        whenever(wrapped.bulkWrite(clientSession.wrapped, requests)).doReturn(mock())
+        whenever(wrapped.bulkWrite(clientSession.wrapped, requests, options)).doReturn(mock())
+
+        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)
+    }
 }
diff --git a/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java b/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java
index 1b7c3a40716..8141af36918 100644
--- a/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java
+++ b/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java
@@ -403,18 +403,6 @@ private void assertOperation(final UnifiedTestContext context, final BsonDocumen
 
     private static void assertOperationResult(final UnifiedTestContext context, final BsonDocument operation, final int operationIndex,
             final OperationResult result) {
-        if (result.getException() instanceof org.opentest4j.TestAbortedException) {
-            // BULK-TODO remove
-            if (result.getException().getMessage().contains("BULK-TODO Kotlin implement")) {
-                throw (org.opentest4j.TestAbortedException) result.getException();
-            }
-        }
-        if (result.getException() instanceof org.junit.AssumptionViolatedException) {
-            // BULK-TODO remove
-            if (result.getException().getMessage().contains("BULK-TODO Kotlin implement")) {
-                throw (org.junit.AssumptionViolatedException) result.getException();
-            }
-        }
         context.getAssertionContext().push(ContextElement.ofCompletedOperation(operation, result, operationIndex));
 
         if (!operation.getBoolean("ignoreResultAndError", BsonBoolean.FALSE).getValue()) {