What causes "error bulk writing to MongoDB: %w" in apache/beam?

Inspect the wrapped error (%w) for the BulkWriteError index/details to find which document failed and why

error bulk writing to MongoDB

Error message

error bulk writing to MongoDB: %w

What it means

writeFn.flush performs a MongoDB BulkWrite of all accumulated write models using the configured ordered/unordered mode. When the server or driver rejects any part of the bulk operation, the underlying error is wrapped with this message and returned to the Beam runner, failing the bundle. It indicates the batch of documents could not be persisted to the collection.

Solutions

Inspect the wrapped error (%w) for the BulkWriteError index/details to find which document failed and why
Fix the offending documents: resolve duplicate keys, satisfy schema validators, or correct types
Set Ordered=false if you want the bulk write to skip failing documents and apply the rest
Verify MongoDB connectivity, credentials, and that the target collection/database exist
Add retry logic around the pipeline or re-run the failing bundle; ensure fn.models accumulation doesn't grow past server limits

Example fix

// before
collection.BulkWrite(ctx, fn.models, opts) // ordered: one dup key aborts whole batch
// after
opts := options.BulkWrite().SetOrdered(false) // continue past per-document failures
if _, err := fn.collection.BulkWrite(ctx, fn.models, opts); err != nil {
    var bwe mongo.BulkWriteException
    if errors.As(err, &bwe) { /* log per-entry write errors */ }
    return fmt.Errorf("error bulk writing to MongoDB: %w", err)
}

Defensive patterns

Strategy: retry

Validate before calling

// before running the pipeline
if collection == nil || db == nil { return errors.New("mongodb client/collection not initialized") }
for _, m := range models { if m == nil { return errors.New("nil write model in batch") } }

Type guard

func isBulkWriteError(err error) (mongo.BulkWriteException, bool) {
	var bwe mongo.BulkWriteException
	if errors.As(err, &bwe) {
		return bwe, true
	}
	return mongo.BulkWriteException{}, false
}

Try / catch

if _, err := fn.collection.BulkWrite(ctx, fn.models, opts); err != nil {
	var bwe mongo.BulkWriteException
	if errors.As(err, &bwe) {
		for _, we := range bwe.WriteErrors {
			log.Printf("doc %d failed: code=%d msg=%s", we.Index, we.Code, we.Message)
		}
	}
	if isTransient(err) { /* retry with backoff */ }
	return fmt.Errorf("error bulk writing to MongoDB: %w", err)
}

Prevention

Add unique indexes deliberately and deduplicate input keys before writing
Use SetOrdered(false) when partial success is acceptable
Enable MongoDB schema validation in staging matching production
Monitor connection health (ping) and set sane client timeouts/retryWrites
Cap the batch size (len(fn.models)) to avoid oversized bulk requests

When it happens

Trigger: fn.collection.BulkWrite(ctx, fn.models, opts) returns a non-nil error: connection loss mid-flush, document failing schema/validation, duplicate key in an ordered bulk write, write concern timeout, or an empty/invalid collection handle from a misconfigured client.

Common situations: Network partition between the Beam worker and MongoDB; inserting documents that violate a unique index (duplicate key E11000); MongoDB validator rejecting documents; collection dropped while pipeline is running; wrong URI/credentials so the collection handle is unusable.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/d00699e324e7720d. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/io/mongodbio/write.go:198

	emit(key)

	return nil
}

func (fn *writeFn) FinishBundle(ctx context.Context, _ func(beam.X)) error {
	if len(fn.models) > 0 {
		return fn.flush(ctx)
	}

	return nil
}

func (fn *writeFn) flush(ctx context.Context) error {
	opts := options.BulkWrite().SetOrdered(fn.Ordered)

	if _, err := fn.collection.BulkWrite(ctx, fn.models, opts); err != nil {
		return fmt.Errorf("error bulk writing to MongoDB: %w", err)
	}

	fn.models = nil

	return nil
}

View on GitHub (pinned to 12126d8942)