{"record":{"id":"8cf6b2b336428ba2","repo":"risingwavelabs/risingwave","slug":"bulk-write-respond-with-write-errors","errorCode":null,"errorMessage":"bulk write respond with write errors: {:?}","messagePattern":"bulk write respond with write errors: (.+?)","errorType":"exception","errorClass":"SinkError::Mongodb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":93,"sourceCode":"        }\n    }\n\n    async fn send_bulk_write_command(db: Database, command: Document) -> Result<()> {\n        let result = db.run_command(command).await.map_err(|err| {\n            SinkError::Mongodb(anyhow!(err).context(format!(\n                \"sending bulk write command failed, database: {}\",\n                db.name()\n            )))\n        })?;\n\n        if let Ok(ok) = result.get_i32(\"ok\")\n            && ok != 1\n        {\n            return Err(SinkError::Mongodb(anyhow!(\"bulk write write errors\")));\n        }\n\n        if let Ok(write_errors) = result.get_array(\"writeErrors\") {\n            return Err(SinkError::Mongodb(anyhow!(\n                \"bulk write respond with write errors: {:?}\",\n                write_errors,\n            )));\n        }\n\n        if let Ok(write_concern_error) = result.get_array(\"writeConcernError\") {\n            return Err(SinkError::Mongodb(anyhow!(\n                \"bulk write respond with write errors: {:?}\",\n                write_concern_error,\n            )));\n        }\n\n        Ok(())\n    }\n}\n\npub const MONGODB_SINK: &str = \"mongodb\";\nconst MONGODB_SEND_FUTURE_BUFFER_MAX_SIZE: usize = 4096;","sourceCodeStart":75,"sourceCodeEnd":111,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L75-L111","documentation":"The bulk write command succeeded at the command level, but the response contains a non-empty 'writeErrors' array: individual document writes failed (e.g. duplicate key, document validation). The sink reports the write errors back wrapped in a Mongodb SinkError.","triggerScenarios":"result.get_array(\"writeErrors\") returns Ok with entries: per-document failures such as duplicate _id (E11000), document too large (BSONObjectTooLarge), or schema validation failure on the collection.","commonSituations":"Upserting rows whose keys already exist with conflicting values, writing documents exceeding 16MB, collection JSON schema validators rejecting documents, or concurrent writers inserting the same _id.","solutions":["Read the writeErrors entries in the message; fix the flagged documents (duplicate _id, size, validator).","Use upsert behavior in the sink for keys that legitimately collide instead of plain inserts.","Relax or correct the collection's JSON schema validator if valid documents are rejected.","Split oversized documents or widen the schema if the BSON size limit is hit."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// validate document shape/size against collection constraints before bulk write\nif doc.len() > 15 * 1024 * 1024 { return Err(\"document too large\"); }\n","typeGuard":null,"tryCatchPattern":"// parse writeErrors and act per error code (e.g. E11000 duplicate key)\nlet codes: Vec<_> = write_errors.iter().filter_map(|e| e.get_document(&Ok(Default::default())).ok()).collect();\nif codes.iter().any(|d| d.get_i32(\"code\") == Ok(11000)) { enable_upsert_or_dedup(); }","preventionTips":["Use upsert semantics when duplicate keys are expected.","Keep documents well under the 16MB BSON limit.","Align collection JSON schema validators with the sink schema.","Deduplicate _id values upstream before bulk writes."],"tags":["rust","mongodb","sink","bulk-write"],"backgroundTag":"database-write-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}