apache/beam · error

while executing FinishBundle for

Error message

while executing FinishBundle for %v

What it means

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.

Solutions

1. Inspect the wrapped inner error to identify which sink/unit failed finalization
2. Fix the underlying finalize failure (permissions, disk space, destination connectivity)
3. Add buffering/flush error handling in custom sinks so FinishBundle errors are actionable
4. Re-execute with a fresh Plan; the current plan is Broken and cannot be reused

Example fix

// before: sink silently buffers, fails only at FinishBundle
func (s *mySink) FinishBundle(ctx context.Context) error {
    return s.file.Sync()
}
// after: flush eagerly and surface errors during Process
func (s *mySink) ProcessElement(ctx context.Context, elm []byte) error {
    _, err := s.w.Write(elm)
    return err
}
func (s *mySink) FinishBundle(ctx context.Context) error {
    if err := s.w.Flush(); err != nil {
        return fmt.Errorf("flush failed: %w", err)
    }
    return s.file.Sync()
}

// Ensure sinks flush during ProcessElement, not only FinishBundle
_, err := writer.Write(elm); return err

func (s *mySink) isOpen() bool { return s.w != nil }

if err := plan.Execute(ctx); err != nil {
    if strings.Contains(err.Error(), "FinishBundle") {
        log.Printf("finalize failure: %v", errors.Unwrap(err))
    }
    plan.Down()
    return err
}

When it happens

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

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

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• plan failed
• Plan failed callbacks
• plan failed with multiple errors
• while executing Process for
• AfterProcessingTime trigger set without a delay or…

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/plan.go:162

		if err := callNoPanic(ctx, func(ctx context.Context) error { return root.StartBundle(ctx, id, manager) }); err != nil {
			p.setStatus(Broken)
			return errors.Wrapf(err, "while executing StartBundle for %v", p)
		}
	}
	for _, root := range p.roots {
		if err := callNoPanic(ctx, func(ctx context.Context) error {
			cps, err := root.Process(ctx)
			p.checkpoints = cps
			return err
		}); err != nil {
			p.setStatus(Broken)
			return errors.Wrapf(err, "while executing Process for %v", p)
		}
	}
	for _, root := range p.roots {
		if err := callNoPanic(ctx, root.FinishBundle); err != nil {
			p.setStatus(Broken)
			return errors.Wrapf(err, "while executing FinishBundle for %v", p)
		}
	}
	p.setStatus(Up)
	return nil
}

// Finalize runs any callbacks registered by the bundleFinalizer. Should be run on bundle finalization.
func (p *Plan) Finalize() error {
	if s := p.getStatus(); s != Up {
		return errors.Errorf("invalid status for plan %v: %v", p.id, s)
	}
	failedIndices := []int{}
	for idx, bfc := range p.bf.callbacks {
		if time.Now().Before(bfc.validUntil) {
			if err := bfc.callback(); err != nil {
				failedIndices = append(failedIndices, idx)
			}
		}

View on GitHub (pinned to 12126d8942)