{"record":{"id":"53dc17f13376b560","repo":"risingwavelabs/risingwave","slug":"sending-bulk-write-command-failed-database","errorCode":null,"errorMessage":"sending bulk write command failed, database: {}","messagePattern":"sending bulk write command failed, database: (.+?)","errorType":"exception","errorClass":"SinkError::Mongodb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mongodb.rs","lineNumber":80,"sourceCode":"    pub(super) fn send_bulk_write_commands(\n        db: Database,\n        upsert: Option<Document>,\n        delete: Option<Document>,\n    ) -> SendBulkWriteCommandFuture {\n        async move {\n            if let Some(upsert) = upsert {\n                send_bulk_write_command(db.clone(), upsert).await?;\n            }\n            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","sourceCodeStart":62,"sourceCodeEnd":98,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mongodb.rs#L62-L98","documentation":"The MongoDB sink's send_bulk_write_command failed to execute the bulk write command document against the database: db.run_command returned an error (transport/connection/driver level). The error is wrapped with the target database name for context.","triggerScenarios":"db.run_command(command) returns Err: connection dropped, socket timeout, authentication failure, or the MongoDB server rejected the command at the driver level before a reply document was produced.","commonSituations":"MongoDB unreachable or restarting, wrong URI/credentials in the sink config, network partition or DNS failure, server version too old for the command, or idle connection reaped by a load balancer.","solutions":["Verify the MongoDB URI and credentials by connecting with mongosh using the same settings.","Check network reachability to the MongoDB host/port and any load balancer idle-timeout settings.","Increase driver timeouts / enable retryable writes in the connection options.","Confirm the server version supports the bulkWrite command used by the sink.","Retry the sink operation if the failure was a transient network blip."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// pre-check MongoDB connectivity with the same URI\nmongosh \"$MONGO_URI\" --eval 'db.runCommand({ping: 1})'\n","typeGuard":null,"tryCatchPattern":"// treat transport errors as retryable with capped backoff\nmatch classify(&err) {\n    Kind::Network | Kind::Timeout => retry_with_backoff(max_retries = 3),\n    Kind::Auth => alert_and_fail(),\n    _ => alert_and_fail(),\n}","preventionTips":["Verify the MongoDB URI and credentials before creating the sink.","Enable retryable writes and sensible timeouts in connection options.","Set load-balancer idle timeouts longer than write intervals.","Monitor MongoDB server uptime/restarts feeding the sink."],"tags":["rust","mongodb","sink","network"],"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"}