{"record":{"id":"023ddce87fd9bf50","repo":"risingwavelabs/risingwave","slug":"bulk-write-write-errors","errorCode":null,"errorMessage":"bulk write write errors","messagePattern":"bulk write write errors","errorType":"exception","errorClass":"SinkError::Mongodb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":89,"sourceCode":"            if let Some(delete) = delete {\n                send_bulk_write_command(db, delete).await?;\n            }\n            Ok(())\n        }\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    }","sourceCodeStart":71,"sourceCodeEnd":107,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L71-L107","documentation":"After running the bulk write command, the response document's 'ok' field was present but not equal to 1, meaning MongoDB reports the command itself did not succeed. The sink returns a generic 'bulk write write errors' Mongodb error.","triggerScenarios":"result.get_i32(\"ok\") yields ok != 1: the server rejected the bulk write command, e.g. unauthorized, command not supported, or an internal server error returned without a detailed writeErrors array.","commonSituations":"Sink user lacking write role on the target database, mongos/sharded cluster rejecting the command, or server-side errors like command-not-allowed on restricted deployments (e.g. Atlas free tier limits).","solutions":["Grant the sink user write privileges (dbAdmin/write role) on the target database.","Check the server's full response by enabling driver command logging to see the actual errmsg.","Verify the MongoDB deployment supports the bulk write command (version/compatibility).","Retry if the server reported a transient internal error."],"exampleFix":"// before: generic error loses the server message\nreturn Err(SinkError::Mongodb(anyhow!(\"bulk write write errors\")));\n// after: include server errmsg for diagnosability\nlet msg = result.get_str(\"errmsg\").unwrap_or(\"unknown\");\nreturn Err(SinkError::Mongodb(anyhow!(\"bulk write failed: ok != 1 ({msg})\")));","handlingStrategy":"validation","validationCode":"// verify sink user permissions before writing\nmongosh \"$MONGO_URI\" --eval \"db.runCommand({connectionStatus: 1, showPrivileges: true})\"\n","typeGuard":null,"tryCatchPattern":"// surface errmsg from the response instead of failing blind\nif ok != 1 {\n    let msg = result.get_str(\"errmsg\").unwrap_or(\"unknown\");\n    return Err(SinkError::Mongodb(anyhow!(\"bulk write failed: {msg}\")));\n}","preventionTips":["Grant the sink user write roles on the target database.","Confirm the deployment supports the bulkWrite command (server version check).","Test the same command via mongosh as the sink user.","Review sharded-cluster/Atlas tier restrictions on write commands."],"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"}