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.

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)

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

• SinkError::SqlServer(anyhow!(err))
• bulk write write errors
• bulk write respond with write errors: {:?}
• failed to send hello command to mongodb
• {batch write failure, with context()}

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