What causes "invalid status for pardo %v: %v, want Initializing" in apache/beam?

Ensure Up() is called exactly once per ParDo instance; construct a fresh Plan via exec.NewPlan for each retry.

invalid status for pardo %v: %v, want Initializing

Error message

invalid status for pardo %v: %v, want Initializing

What it means

ParDo.Up() initializes a ParDo execution unit and requires it to be in the Initializing status; the library enforces a strict lifecycle status machine (Initializing -> Up -> Active -> Up ...). This error is thrown when Up() is called on a ParDo whose status is anything other than Initializing, which means the unit was already initialized or was never in a fresh state.

Source

Thrown at sdks/go/pkg/beam/core/runtime/exec/pardo.go:91

	key       typex.Window
	sideinput []ReusableInput
	extra     []any
}

// ID returns the UnitID for this ParDo.
func (n *ParDo) ID() UnitID {
	return n.UID
}

// HasOnTimer returns if this ParDo wraps a DoFn that has an OnTimer method.
func (n *ParDo) HasOnTimer() bool {
	return n.TimerTracker != nil
}

// Up initializes this ParDo and does one-time DoFn setup.
func (n *ParDo) Up(ctx context.Context) error {
	if n.status != Initializing {
		return errors.Errorf("invalid status for pardo %v: %v, want Initializing", n.UID, n.status)
	}
	n.status = Up
	n.inv = newInvoker(n.Fn.ProcessElementFn())
	if fn, ok := n.Fn.OnTimerFn(); ok {
		n.onTimerInvoker = newInvoker(fn)
	}

	n.states = metrics.NewPTransformState(n.PID)

	// We can't cache the context during Setup since it runs only once per bundle.
	// Subsequent bundles might run this same node, and the context here would be
	// incorrectly refering to the older bundleId.
	setupCtx := metrics.SetPTransformID(ctx, n.PID)
	if _, err := InvokeWithOptsWithoutEventTime(setupCtx, n.Fn.SetupFn(), InvokeOpts{}); err != nil {
		return n.fail(err)
	}

	emitters, err := makeEmitters(n.Fn.ProcessElementFn(), n.Out)

View on GitHub (pinned to 12126d8942)

Solutions

Ensure Up() is called exactly once per ParDo instance; construct a fresh Plan via exec.NewPlan for each retry.
Check plan status before calling Up; Plan.Execute already handles the Initializing check, so prefer calling Plan.Execute instead of Up directly.
If the unit is Broken due to an earlier error, rebuild the pipeline and construct new units instead of reusing them.

Example fix

// before
for _, u := range units {
    u.Up(ctx)
}
// after (let Plan.Execute manage the lifecycle once)
plan, err := exec.NewPlan("plan-id", units)
if err != nil { return err }
err = plan.Execute(ctx, bundleID, dataContext)

Defensive patterns

Strategy: validation

Validate before calling

if pardo.status != exec.Initializing {
    return fmt.Errorf("pardo %v already initialized (status %v)", pardo.UID, pardo.status)
}
err := pardo.Up(ctx)

Try / catch

if err := plan.Execute(ctx, id, mgr); err != nil {
    if strings.Contains(err.Error(), "invalid status for pardo") {
        // rebuild a fresh plan instead of retrying
        plan, err = exec.NewPlan(id, freshUnits())
    }
}

Prevention

Never call unit-level Up directly; rely on Plan.Execute for lifecycle management.
Create a new plan per bundle/retry instead of reusing units.
Track unit status in custom runner code before invoking lifecycle methods.

When it happens

Trigger: Calling ParDo.Up(ctx) twice on the same unit, reusing a Plan after it has been Up'd, or Up'ing a unit extracted from a plan that has already executed.

Common situations: Developers manually constructing exec plans in tests or custom runners and calling Up more than once; retry logic that re-runs plan initialization on the same plan object after a failure.

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

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