{"record":{"id":"fe322a330bd368fc","repo":"apache/beam","slug":"while-executing-finishbundle-for-v","errorCode":null,"errorMessage":"while executing FinishBundle for %v","messagePattern":"while executing FinishBundle for (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/plan.go","lineNumber":162,"sourceCode":"\t\tif err := callNoPanic(ctx, func(ctx context.Context) error { return root.StartBundle(ctx, id, manager) }); err != nil {\n\t\t\tp.setStatus(Broken)\n\t\t\treturn errors.Wrapf(err, \"while executing StartBundle for %v\", p)\n\t\t}\n\t}\n\tfor _, root := range p.roots {\n\t\tif err := callNoPanic(ctx, func(ctx context.Context) error {\n\t\t\tcps, err := root.Process(ctx)\n\t\t\tp.checkpoints = cps\n\t\t\treturn err\n\t\t}); err != nil {\n\t\t\tp.setStatus(Broken)\n\t\t\treturn errors.Wrapf(err, \"while executing Process for %v\", p)\n\t\t}\n\t}\n\tfor _, root := range p.roots {\n\t\tif err := callNoPanic(ctx, root.FinishBundle); err != nil {\n\t\t\tp.setStatus(Broken)\n\t\t\treturn errors.Wrapf(err, \"while executing FinishBundle for %v\", p)\n\t\t}\n\t}\n\tp.setStatus(Up)\n\treturn nil\n}\n\n// Finalize runs any callbacks registered by the bundleFinalizer. Should be run on bundle finalization.\nfunc (p *Plan) Finalize() error {\n\tif s := p.getStatus(); s != Up {\n\t\treturn errors.Errorf(\"invalid status for plan %v: %v\", p.id, s)\n\t}\n\tfailedIndices := []int{}\n\tfor idx, bfc := range p.bf.callbacks {\n\t\tif time.Now().Before(bfc.validUntil) {\n\t\t\tif err := bfc.callback(); err != nil {\n\t\t\t\tfailedIndices = append(failedIndices, idx)\n\t\t\t}\n\t\t}","sourceCodeStart":144,"sourceCodeEnd":180,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/plan.go#L144-L180","documentation":"During Plan.Execute, after all Process() calls succeed, each root unit's FinishBundle() is invoked to run bundle-finalization logic (flushing buffers, sinks, metric commits). If FinishBundle returns an error or panics, the plan is marked Broken and the error is wrapped with the plan identity.","triggerScenarios":"Any root unit's FinishBundle() returns a non-nil error during Plan.Execute — typically a sink's finalize (e.g. file close/rename, database flush) or DoFn FinishBundle hook failing.","commonSituations":"Sink finalization fails (cannot close/rename output file, permission issue on destination, S3/GCS flush failure), a buffered DoFn returns an error in FinishBundle, or state accumulated during Process makes finalize fail.","solutions":["Inspect the wrapped inner error to identify which sink/unit failed finalization","Fix the underlying finalize failure (permissions, disk space, destination connectivity)","Add buffering/flush error handling in custom sinks so FinishBundle errors are actionable","Re-execute with a fresh Plan; the current plan is Broken and cannot be reused"],"exampleFix":"// before: sink silently buffers, fails only at FinishBundle\nfunc (s *mySink) FinishBundle(ctx context.Context) error {\n    return s.file.Sync()\n}\n// after: flush eagerly and surface errors during Process\nfunc (s *mySink) ProcessElement(ctx context.Context, elm []byte) error {\n    _, err := s.w.Write(elm)\n    return err\n}\nfunc (s *mySink) FinishBundle(ctx context.Context) error {\n    if err := s.w.Flush(); err != nil {\n        return fmt.Errorf(\"flush failed: %w\", err)\n    }\n    return s.file.Sync()\n}","handlingStrategy":"try-catch","validationCode":"// Ensure sinks flush during ProcessElement, not only FinishBundle\n_, err := writer.Write(elm); return err","typeGuard":"func (s *mySink) isOpen() bool { return s.w != nil }","tryCatchPattern":"if err := plan.Execute(ctx); err != nil {\n    if strings.Contains(err.Error(), \"FinishBundle\") {\n        log.Printf(\"finalize failure: %v\", errors.Unwrap(err))\n    }\n    plan.Down()\n    return err\n}","preventionTips":["Flush buffers incrementally in ProcessElement rather than relying on FinishBundle","Make FinishBundle idempotent and tolerant of already-flushed state","Verify sink destination permissions/connectivity before bundle execution","Keep FinishBundle error messages descriptive enough to identify the unit"],"tags":["go","apache-beam","execution-plan","finish-bundle"],"backgroundTag":"invalid-state-transition","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"}