risingwavelabs/risingwave · error · SinkError::Mongodb

sending bulk write command failed, database

Error message

sending bulk write command failed, database: {}

What it means

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.

Solutions

1. Verify the MongoDB URI and credentials by connecting with mongosh using the same settings.
2. Check network reachability to the MongoDB host/port and any load balancer idle-timeout settings.
3. Increase driver timeouts / enable retryable writes in the connection options.
4. Confirm the server version supports the bulkWrite command used by the sink.
5. Retry the sink operation if the failure was a transient network blip.

// pre-check MongoDB connectivity with the same URI
mongosh "$MONGO_URI" --eval 'db.runCommand({ping: 1})'

// treat transport errors as retryable with capped backoff
match classify(&err) {
    Kind::Network | Kind::Timeout => retry_with_backoff(max_retries = 3),
    Kind::Auth => alert_and_fail(),
    _ => alert_and_fail(),
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• {batch write failure, with context()}
• bulk write respond with write errors
• bulk write write errors
• {connection_err (pulsar::Error) after retries exhausted}
• failed to send hello command to mongodb

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/53dc17f13376b560. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/mongodb.rs:80

    pub(super) fn send_bulk_write_commands(
        db: Database,
        upsert: Option<Document>,
        delete: Option<Document>,
    ) -> SendBulkWriteCommandFuture {
        async move {
            if let Some(upsert) = upsert {
                send_bulk_write_command(db.clone(), upsert).await?;
            }
            if let Some(delete) = delete {
                send_bulk_write_command(db, delete).await?;
            }
            Ok(())
        }
    }

    async fn send_bulk_write_command(db: Database, command: Document) -> Result<()> {
        let result = db.run_command(command).await.map_err(|err| {
            SinkError::Mongodb(anyhow!(err).context(format!(
                "sending bulk write command failed, database: {}",
                db.name()
            )))
        })?;

        if let Ok(ok) = result.get_i32("ok")
            && ok != 1
        {
            return Err(SinkError::Mongodb(anyhow!("bulk write write errors")));
        }

        if let Ok(write_errors) = result.get_array("writeErrors") {
            return Err(SinkError::Mongodb(anyhow!(
                "bulk write respond with write errors: {:?}",
                write_errors,
            )));
        }

View on GitHub (pinned to 6469eb736d)