apache/beam · error

invalid status for combine

Error message

invalid status for combine %v: %v

What it means

Combine.Up enforces the lifecycle state machine: the node must be in the Initializing status when Up is called. Calling Up on an already-initialized Combine (status Up, Active, etc.) indicates the bundle lifecycle methods are being invoked out of order or more than once.

Solutions

1. Call each lifecycle method exactly once and in order: Up, StartBundle, ProcessElement(s), FinishBundle.
2. Reset status to Initializing (construct a fresh Combine) before calling Up again.
3. Check for runner code paths that re-invoke Up on a shared plan instance.

// guard before calling Up
// Combine has no exported status field; drive lifecycle through plan.Execute instead of direct calls.

if err := combine.Up(ctx); err != nil {
	if strings.Contains(err.Error(), "invalid status for combine") {
		// the node was already initialized; rebuild the plan instead of re-upping
	}
	return err
}

When it happens

Trigger: Calling Up twice on the same Combine unit, or calling Up after StartBundle/FinishBundle without re-initialization.

Common situations: Custom runner code or tests building a plan and invoking lifecycle methods manually; runner bugs that re-up a cached plan without resetting node status.

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

• invalid status for combine convert
• invalid status for combine extract
• invalid status for combine merge
• invalid status for precombine
• instruction no longer processing

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/combine.go:71

	aiValConvert func(any) any

	states *metrics.PTransformState
}

// GetPID returns the PTransformID for this CombineFn.
func (n *Combine) GetPID() string {
	return n.PID
}

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

// Up initializes this CombineFn and runs its SetupFn() method.
func (n *Combine) Up(ctx context.Context) error {
	if n.status != Initializing {
		return errors.Errorf("invalid status for combine %v: %v", n.UID, n.status)
	}
	n.status = Up

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

	if _, err := InvokeWithoutEventTime(ctx, n.Fn.SetupFn(), nil, nil, nil, nil, nil); err != nil {
		return n.fail(err)
	}

	if ca := n.Fn.CreateAccumulatorFn(); ca != nil {
		n.createAccumInv = newInvoker(ca)
	}
	if ai := n.Fn.AddInputFn(); ai != nil {
		n.addInputInv = newInvoker(ai)
	} else {
		n.optimizeMergeFn()
	}
	n.mergeInv = newInvoker(n.Fn.MergeAccumulatorsFn())

View on GitHub (pinned to 12126d8942)