{"record":{"id":"55b10c253555b2a3","repo":"apache/beam","slug":"invalid-status-for-pardo-v-v-want-initializing","errorCode":null,"errorMessage":"invalid status for pardo %v: %v, want Initializing","messagePattern":"invalid status for pardo (.+?): (.+?), want Initializing","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/pardo.go","lineNumber":91,"sourceCode":"\tkey       typex.Window\n\tsideinput []ReusableInput\n\textra     []any\n}\n\n// ID returns the UnitID for this ParDo.\nfunc (n *ParDo) ID() UnitID {\n\treturn n.UID\n}\n\n// HasOnTimer returns if this ParDo wraps a DoFn that has an OnTimer method.\nfunc (n *ParDo) HasOnTimer() bool {\n\treturn n.TimerTracker != nil\n}\n\n// Up initializes this ParDo and does one-time DoFn setup.\nfunc (n *ParDo) Up(ctx context.Context) error {\n\tif n.status != Initializing {\n\t\treturn errors.Errorf(\"invalid status for pardo %v: %v, want Initializing\", n.UID, n.status)\n\t}\n\tn.status = Up\n\tn.inv = newInvoker(n.Fn.ProcessElementFn())\n\tif fn, ok := n.Fn.OnTimerFn(); ok {\n\t\tn.onTimerInvoker = newInvoker(fn)\n\t}\n\n\tn.states = metrics.NewPTransformState(n.PID)\n\n\t// We can't cache the context during Setup since it runs only once per bundle.\n\t// Subsequent bundles might run this same node, and the context here would be\n\t// incorrectly refering to the older bundleId.\n\tsetupCtx := metrics.SetPTransformID(ctx, n.PID)\n\tif _, err := InvokeWithOptsWithoutEventTime(setupCtx, n.Fn.SetupFn(), InvokeOpts{}); err != nil {\n\t\treturn n.fail(err)\n\t}\n\n\temitters, err := makeEmitters(n.Fn.ProcessElementFn(), n.Out)","sourceCodeStart":73,"sourceCodeEnd":109,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/pardo.go#L73-L109","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nfor _, u := range units {\n    u.Up(ctx)\n}\n// after (let Plan.Execute manage the lifecycle once)\nplan, err := exec.NewPlan(\"plan-id\", units)\nif err != nil { return err }\nerr = plan.Execute(ctx, bundleID, dataContext)","handlingStrategy":"validation","validationCode":"if pardo.status != exec.Initializing {\n    return fmt.Errorf(\"pardo %v already initialized (status %v)\", pardo.UID, pardo.status)\n}\nerr := pardo.Up(ctx)","typeGuard":null,"tryCatchPattern":"if err := plan.Execute(ctx, id, mgr); err != nil {\n    if strings.Contains(err.Error(), \"invalid status for pardo\") {\n        // rebuild a fresh plan instead of retrying\n        plan, err = exec.NewPlan(id, freshUnits())\n    }\n}","preventionTips":["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."],"tags":["go","apache-beam","lifecycle","state-machine"],"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-14T16:17:12.679Z"}