apache/beam · error

plan failed

Error message

plan %v failed

What it means

Plan.Down() tears down all units and collects any errors from unit Down() calls. If exactly one unit failed teardown, that error is wrapped with "plan %v failed"; the plan's teardown therefore did not fully complete.

Solutions

1. Inspect the wrapped error to identify which unit's Down() failed
2. Fix the failing unit's cleanup logic (handle double-close, tolerate already-released resources)
3. Make custom sink/DoFn Down() methods idempotent so repeated teardown doesn't error
4. Ensure process-level resources (files, connections) are closed even if Down() errors, to avoid leaks

Example fix

// before: Down fails on double-close
func (s *mySink) Down(ctx context.Context) error {
    return s.file.Close()
}
// after
func (s *mySink) Down(ctx context.Context) error {
    if s.file == nil {
        return nil
    }
    err := s.file.Close()
    s.file = nil
    return err
}

// Make unit cleanup safe before teardown
func (s *mySink) Down(ctx context.Context) error {
    if s.file == nil { return nil }
    return s.file.Close()
}

func (s *mySink) isClosed() bool { return s.file == nil }

if err := plan.Down(); err != nil {
    log.Printf("teardown error (check wrapped cause): %v", errors.Unwrap(err))
    // attempt process-level resource cleanup as fallback
}

When it happens

Trigger: Calling Plan.Down() when exactly one of the plan's units returns a non-nil error from its Down() method (e.g. a sink failing to close resources, a DoFn failing cleanup).

Common situations: Worker shutdown with a custom sink whose Close() fails (already-closed file, network flush failure); datasource failing to release a reader; resource leaks accumulating when the same error recurs at every bundle teardown.

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 with multiple errors
• Plan failed callbacks
• while executing FinishBundle for
• 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/8cc20e2ea7392dac. Report an issue: GitHub.

Appendix: source

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

func (p *Plan) Down(ctx context.Context) error {
	// Technically racy, but only one thread calls this method on the plan.
	if p.getStatus() == Down {
		return nil // ok: already down
	}
	p.setStatus(Down)

	var errs []error
	for _, u := range p.units {
		if err := callNoPanic(ctx, u.Down); err != nil {
			errs = append(errs, err)
		}
	}

	switch len(errs) {
	case 0:
		return nil
	case 1:
		return errors.Wrapf(errs[0], "plan %v failed", p.id)
	default:
		return errors.Errorf("plan %v failed with multiple errors: %v", p.id, errs)
	}
}

func (p *Plan) String() string {
	var units []string
	for i := len(p.units) - 1; i >= 0; i-- {
		u := p.units[i]
		units = append(units, fmt.Sprintf("%v: %v", u.ID(), u))
	}
	return fmt.Sprintf("Plan[%v]:\n%v", p.ID(), strings.Join(units, "\n"))
}

// PlanSnapshot contains system metrics for the current run of the plan.
type PlanSnapshot struct {
	Source ProgressReportSnapshot
	PCols  []PCollectionSnapshot

View on GitHub (pinned to 12126d8942)