{"record":{"id":"d00699e324e7720d","repo":"apache/beam","slug":"error-bulk-writing-to-mongodb-w","errorCode":null,"errorMessage":"error bulk writing to MongoDB: %w","messagePattern":"error bulk writing to MongoDB: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/io/mongodbio/write.go","lineNumber":198,"sourceCode":"\n\temit(key)\n\n\treturn nil\n}\n\nfunc (fn *writeFn) FinishBundle(ctx context.Context, _ func(beam.X)) error {\n\tif len(fn.models) > 0 {\n\t\treturn fn.flush(ctx)\n\t}\n\n\treturn nil\n}\n\nfunc (fn *writeFn) flush(ctx context.Context) error {\n\topts := options.BulkWrite().SetOrdered(fn.Ordered)\n\n\tif _, err := fn.collection.BulkWrite(ctx, fn.models, opts); err != nil {\n\t\treturn fmt.Errorf(\"error bulk writing to MongoDB: %w\", err)\n\t}\n\n\tfn.models = nil\n\n\treturn nil\n}\n","sourceCodeStart":180,"sourceCodeEnd":205,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/io/mongodbio/write.go#L180-L205","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\ncollection.BulkWrite(ctx, fn.models, opts) // ordered: one dup key aborts whole batch\n// after\nopts := options.BulkWrite().SetOrdered(false) // continue past per-document failures\nif _, err := fn.collection.BulkWrite(ctx, fn.models, opts); err != nil {\n    var bwe mongo.BulkWriteException\n    if errors.As(err, &bwe) { /* log per-entry write errors */ }\n    return fmt.Errorf(\"error bulk writing to MongoDB: %w\", err)\n}","handlingStrategy":"retry","validationCode":"// before running the pipeline\nif collection == nil || db == nil { return errors.New(\"mongodb client/collection not initialized\") }\nfor _, m := range models { if m == nil { return errors.New(\"nil write model in batch\") } }","typeGuard":"func isBulkWriteError(err error) (mongo.BulkWriteException, bool) {\n\tvar bwe mongo.BulkWriteException\n\tif errors.As(err, &bwe) {\n\t\treturn bwe, true\n\t}\n\treturn mongo.BulkWriteException{}, false\n}","tryCatchPattern":"if _, err := fn.collection.BulkWrite(ctx, fn.models, opts); err != nil {\n\tvar bwe mongo.BulkWriteException\n\tif errors.As(err, &bwe) {\n\t\tfor _, we := range bwe.WriteErrors {\n\t\t\tlog.Printf(\"doc %d failed: code=%d msg=%s\", we.Index, we.Code, we.Message)\n\t\t}\n\t}\n\tif isTransient(err) { /* retry with backoff */ }\n\treturn fmt.Errorf(\"error bulk writing to MongoDB: %w\", err)\n}","preventionTips":["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"],"tags":["mongodb","database","write","beam-io"],"backgroundTag":"database-write-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}