This is part 1/6 of the MongoDB sharded cluster write path series.
Questions this page covers: If you send 1,000 bulk writes to a sharded collection as ORDERED, how many round trips happen between mongos and the shards? Why does one slow shard slow down the whole bulk? What do you gain by switching to UNORDERED, and what do you pay for it?
Reference version: MongoDB 4.2 (source tagr4.2.12, 4.2 official documentation). Support for 4.2 ended on 2023-04-30 (EOL). The mongos behavior of 4.4 and later was not checked (as of 2026-11).
When mongos receives an ORDERED bulk, it sends only a contiguous run of writes bound for the same shard as one batch, and builds the next batch only after it gets that response. A hashed shard key scatters neighboring documents in the bulk across many shards, so the batch is cut almost at every document. With n documents and S shards, the expected number of round trips is 1 + (n − 1)(1 − 1/S). Sending 1,000 documents to 5 shards takes about 800. If the bulk does not depend on order, switch to BulkMode.UNORDERED (driver option ordered(false)), and mongos sends the per-shard batches all at once, concurrently.
A bulk write sends many inserts, updates, replaces, and deletes in one request. In the shell you use db.collection.bulkWrite(). In Spring Data MongoDB you use BulkOperations. The client sends the request once and receives the response once.
A bulk has two modes. The "Ordered vs Unordered Operations" section of the 4.2 official documentation describes them as follows.
| Mode | How it runs | If an error occurs midway |
|---|---|---|
| ordered (default) | "MongoDB executes the operations serially" | Returns without processing the remaining operations |
| unordered | May run in parallel (not guaranteed) | Keeps processing the remaining operations |
In a sharded cluster, the client connects to mongos (the router). mongos looks at the shard key of each write in the bulk and decides which shard it goes to (targeting). It then groups the writes bound for the same shard and sends one child batch per shard.
From the client's point of view, there is one request. Between mongos and the shards, though, there are as many round trips as there are child batches. "Round trip" on this page means this mongos ↔ shard round trip.
A hashed shard key splits chunks (units of data divided by ranges of shard key values) by the hash of the field value. The 4.2 official documentation "Hashed Sharding" says:
Post-hash, documents with "close" shard key values are unlikely to be on the same chunk or shard
For example, take a store_inventory collection that uses _id: { productId, storeId } as a hashed shard key. It is a collection that spreads one product into as many documents as there are stores. The store 1 document and the store 2 document for the same product have similar keys, but after hashing they are likely to sit on different shards. This is the cost a hashed shard key pays in exchange for spreading data evenly.
mongos builds child batches in BatchWriteOp::targetBatch in src/mongo/s/write_ops/batch_write_op.cpp. The comment at lines 263-292 states the ordered rule like this.
// Targeting of ordered batches is a bit more complex - to respect the ordering of the
// batch, we can only send:
// A) a single targeted batch to one shard endpoint
// B) multiple targeted batches, but only containing targeted writes for a single write op
In the usual case where one write goes to only one shard, A applies. In ordered mode, a batch goes to only one shard at a time. If the next write goes to a shard that is not in the current batch, isNewBatchRequiredOrdered (lines 96-106) returns true. mongos then cuts the batch.
The execution loop is in batch_write_exec.cpp. mongos builds and sends batches inside while (!batchOp.isFinished() && !abortBatch) (line 120). It then waits for responses with ars.next(). The comment at line 255 is "Block until a response is available." It builds the next batch only after receiving all responses. On this page, one pass of this loop, from sending a batch to receiving all its responses, is called a round.
So in ordered mode, one contiguous run bound for the same shard is one round trip.
Unordered mode targets all remaining writes and groups them into one batch per shard. In one round it sends up to S child batches (S is the number of shards) concurrently. There are three conditions that cut a batch.
kMaxWriteBatchSize (100,000 writes, write_ops.h:62)BSONObjMaxUserSize)isNewBatchRequiredUnordered, lines 112-125)A 1,000-document bulk usually hits none of these conditions, so it finishes in one round.
Say there are three shards, A, B, and C, and 8 documents are assigned as below. The assignment is an example value.
| Order | Document | Shard |
|---|---|---|
| 1 | d1 | A |
| 2 | d2 | B |
| 3 | d3 | B |
| 4 | d4 | C |
| 5 | d5 | A |
| 6 | d6 | C |
| 7 | d7 | C |
| 8 | d8 | B |
The batches the two modes build are as follows.
| Round | ORDERED | UNORDERED |
|---|---|---|
| 1 | A: [d1] | A: [d1, d5], B: [d2, d3, d8], C: [d4, d6, d7] (sent concurrently) |
| 2 | B: [d2, d3] | None |
| 3 | C: [d4] | None |
| 4 | A: [d5] | None |
| 5 | C: [d6, d7] | None |
| 6 | B: [d8] | None |
ORDERED runs 6 rounds one after another. Each round waits for the response of the previous round. UNORDERED sends 3 child batches concurrently in 1 round.
The number of ORDERED round trips is the number of times the batch is cut, plus 1. The batch is cut when two neighboring documents are on different shards. In the example above, the shard changes at 5 of the 7 neighboring pairs, so there are 1 + 5 = 6 rounds.
Assume a hashed shard key assigns documents to S shards evenly and independently. Then the probability that two neighboring documents are on the same shard is 1/S, and the probability that they differ is 1 − 1/S. With 1,000 documents there are 999 boundaries between documents. With 5 shards, the batch is cut at 4/5 of them, about 799. Add 1 for the first batch and you get about 800.
The general formula is as follows.
기대 왕복 수 = 1 + (n − 1) × (1 − 1/S)
n = 1,000, S = 5 → 1 + 999 × 0.8 = 800.2
The computed values by number of shards are as follows (n = 1,000).
| Number of shards S | Expected ORDERED round trips | UNORDERED rounds (concurrent requests) |
|---|---|---|
| 2 | 500.5 | 1 (2) |
| 3 | 667 | 1 (3) |
| 4 | 750.25 | 1 (4) |
| 5 | 800.2 | 1 (5) |
| 6 | 833.5 | 1 (6) |
| 8 | 875.125 | 1 (8) |
The more shards you add, the closer the number of ORDERED round trips gets to n. If you add shards to raise write throughput, a single ORDERED bulk is actually split into even smaller pieces.
A write ticket is an admission pass that limits how many writes can enter the storage engine at the same time. When a shard's mongod runs a write, it takes the global lock in IX (intent exclusive) mode and gets a write ticket at that point. This behavior is in lock_state.cpp. The function that takes the global lock, _lockGlobalBegin, first calls _acquireTicket. A child batch holds a ticket only while it runs on that shard. It does not hold one while mongos receives the response and builds the next batch.
So an ORDERED bulk does not hold a ticket for long. Instead, the lifetime of one bulk spans every shard. If one shard slows down, that whole bulk stalls. With 5 shards, one bulk passes through each shard 160 times on average (800 ÷ 5). If one shard's responses become 10 times slower, the bulk waits for that slow response 160 times in a row.
By Little's law, the number of jobs in flight in a system equals the number of jobs arriving per second multiplied by the time one job stays in the system. All numbers below are example values.
| Situation | Time for one bulk | Bulks in flight |
|---|---|---|
| ORDERED, normal | 800 × 1ms + 1,000 × 0.2ms = 1.0s | 20 |
| ORDERED, one shard slow | about 2.8s | about 56 |
| UNORDERED, normal | 1ms + 200 × 0.2ms = 41ms | about 0.8 |
| UNORDERED, one shard slow | 10ms + 200 × 2ms = 410ms | about 8 |
The ORDERED, slow row is computed like this. The 160 rounds and 200 documents bound for the slow shard take 1.6s + 0.4s = 2.0s. The remaining 640 rounds and 800 documents take 0.64s + 0.16s = 0.8s.
UNORDERED also waits for the slowest shard. It pays that wait only once, though.
The more bulks are in flight, the more client threads and connections are tied up by the same count. If the writer service that sends the bulks is a Kafka consumer, the consumer thread cannot process the next message during that time.
data class StoreInventory(
@Id val id: InventoryId, // { productId, storeId }, hashed 샤드키
val stock: Int,
val price: Long,
)
fun saveAll(items: List<StoreInventory>) {
val ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, StoreInventory::class.java)
items.forEach { item ->
ops.replaceOne(
Query.query(Criteria.where("_id").`is`(item.id)),
item,
FindAndReplaceOptions.options().upsert(),
)
}
ops.execute()
}
fun saveAll(items: List<StoreInventory>) {
// 같은 _id가 두 번 들어가면 순서가 결과를 바꾼다. 마지막 값만 남긴다.
val deduped = items.associateBy { it.id }.values.toList()
val ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, StoreInventory::class.java)
deduped.forEach { item ->
ops.replaceOne(
Query.query(Criteria.where("_id").`is`(item.id)),
item,
FindAndReplaceOptions.options().upsert(),
)
}
try {
ops.execute()
} catch (e: BulkOperationException) {
val failed = e.errors.map { deduped[it.index] }
retryOrDeadLetter(failed)
}
}
If each replaceOne touches a different document, the execution order does not change the result. So the two pieces of code produce the same final state. Order dependence appears when one bulk contains the same document twice. The UNORDERED code removes that case by keeping only the last value per _id before building the bulk.
The price of switching to UNORDERED is partial failure handling. ORDERED stops at the first error, so how far processing got is fixed at one position. UNORDERED keeps processing the rest even after an error. So failed writes can be scattered throughout the bulk. Each entry in BulkOperationException.errors has an index that tells which position that write had in the bulk. Pick out only the documents at the failed positions and process them again.
If you use the official Java driver directly, one option line makes the switch.
collection.bulkWrite(models, BulkWriteOptions().ordered(false))
_id can repeat, merge them before building the bulk.index entries or send them to a separate queue.db.collection.getShardDistribution() and sh.status(). To see which shard a given key goes to, add explain() to a find() that queries by that key. The shard name in winningPlan.shards of the result is that key's shard. From 4.0 you can also see the hash value with convertShardKeyToHashed().Pitfalls
- The default mode of Spring Data MongoDB
bulkOpsand the shellbulkWrite()is ordered. If you did not specify a mode, it is running as ORDERED.- UNORDERED also waits for the slowest shard. It only cuts the impact of a slow shard down to once; it does not remove it.
- In UNORDERED, if one bulk contains the same document twice, you cannot predict which write ends up applied last.
- The round trips happen between mongos and the shards. So you do not see them if you look only at client-side request counts.
This procedure was not verified by running it. The results below are expectations based on the source.
mlaunch init --replicaset --nodes 1 --sharded 5 with mlaunch (a local cluster launcher included in mtools). A docker compose setup also works.sh.enableSharding("demo"), then sh.shardCollection("demo.store_inventory", { _id: "hashed" }). Sharding an empty collection with a hashed key pre-splits chunks across the shards.db.serverStatus().metrics.commands.update.total._id values via db.store_inventory.bulkWrite(ops, { ordered: true }).{ ordered: false } and repeat steps 3-5.The expected (source-based) results are these. The ORDERED sum of increments lands near the values in the table above. The UNORDERED sum of increments equals the number of shards. This expectation assumes that one child batch arrives at a shard as one update command.
- Actual chunk distribution: The formula above assumes documents are assigned to all shards evenly and independently. If chunks are concentrated on one shard, the chance that the same shard repeats goes up and the number of round trips drops. How much it drops for a given amount of skew was not calculated.
- mongos in 4.4 and later: The basis is the 4.2 source (
r4.2.12) and the 4.2 documentation. Whether the batching rule of mongos in 4.4 and later is the same was not checked.- The counter in the reproduction procedure: Whether
metrics.commands.update.totalrises by exactly 1 per child batch, and whether the profiler records entries per child batch or per write (statement), was not confirmed by running it.- Code examples: The Kotlin examples were written against the Spring Data MongoDB
BulkOperationsAPI but were not run.