risingwavelabs/risingwave · error · SinkError::Mongodb

bulk write respond with write errors: {:?}

Error message

bulk write respond with write errors: {:?}

What it means

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.

Source

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

        }
    }

    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,
            )));
        }

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

        Ok(())
    }
}

pub const MONGODB_SINK: &str = "mongodb";
const MONGODB_SEND_FUTURE_BUFFER_MAX_SIZE: usize = 4096;

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the writeErrors entries in the message; fix the flagged documents (duplicate _id, size, validator).
2. Use upsert behavior in the sink for keys that legitimately collide instead of plain inserts.
3. Relax or correct the collection's JSON schema validator if valid documents are rejected.
4. Split oversized documents or widen the schema if the BSON size limit is hit.

// validate document shape/size against collection constraints before bulk write
if doc.len() > 15 * 1024 * 1024 { return Err("document too large"); }

// parse writeErrors and act per error code (e.g. E11000 duplicate key)
let codes: Vec<_> = write_errors.iter().filter_map(|e| e.get_document(&Ok(Default::default())).ok()).collect();
if codes.iter().any(|d| d.get_i32("code") == Ok(11000)) { enable_upsert_or_dedup(); }

When it happens

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

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

Related errors

• bulk write write errors
• sending bulk write command failed, database: {}
• end of stream
• should have meta client
• should get metadata on checkpoint barrier

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