{"record":{"id":"4a2e71fbae697cff","repo":"apache/beam","slug":"invalid-status-for-pardo-v-v-want-active","errorCode":null,"errorMessage":"invalid status for pardo %v: %v, want Active","messagePattern":"invalid status for pardo (.+?): (.+?), want Active","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/pardo.go","lineNumber":151,"sourceCode":"\n\tn.states.Set(n.ctx, metrics.StartBundle)\n\n\tif err := MultiStartBundle(n.ctx, id, data, n.Out...); err != nil {\n\t\treturn n.fail(err)\n\t}\n\n\t// TODO(BEAM-3303): what to set for StartBundle/FinishBundle window and emitter timestamp?\n\n\tif _, err := n.invokeDataFn(n.ctx, typex.NoFiringPane(), window.SingleGlobalWindow, mtime.ZeroTimestamp, n.Fn.StartBundleFn(), nil); err != nil {\n\t\treturn n.fail(err)\n\t}\n\treturn nil\n}\n\n// ProcessElement processes each parallel element with the DoFn.\nfunc (n *ParDo) ProcessElement(_ context.Context, elm *FullValue, values ...ReStream) error {\n\tif n.status != Active {\n\t\treturn errors.Errorf(\"invalid status for pardo %v: %v, want Active\", n.UID, n.status)\n\t}\n\n\tn.states.Set(n.ctx, metrics.ProcessBundle)\n\n\treturn n.processMainInput(&MainInput{Key: *elm, Values: values})\n}\n\n// processMainInput processes an element that has been converted into a\n// MainInput. Splitting this away from ProcessElement allows other nodes to wrap\n// a ParDo's ProcessElement functionality with their own construction of\n// MainInputs.\nfunc (n *ParDo) processMainInput(mainIn *MainInput) error {\n\tn.TimerTracker.SetCurrentKey(mainIn)\n\telm := &mainIn.Key\n\n\t// If the function observes windows or uses per window state, we must invoke it for each window.\n\t// The expected fast path is that either there is a single window or the function doesn't observe\n\t// windows, so we can optimize it by treating all windows as a single one.","sourceCodeStart":133,"sourceCodeEnd":169,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/pardo.go#L133-L169","documentation":"ParDo.ProcessElement() only processes elements when the unit status is Active, i.e. after a successful StartBundle. The library throws this error when an element arrives outside an active bundle (before StartBundle, after FinishBundle, or when the unit is Broken), because per-element processing depends on the bundle-scoped reader and timer manager.","triggerScenarios":"Feeding elements to ParDo before StartBundle, after FinishBundle, or resuming element processing on a plan whose bundle failed and was marked Broken.","commonSituations":"Split-bundle resumption logic delivering elements to a stale plan; custom direct runners driving the SDF/plan API without the full bundle lifecycle.","solutions":["Call StartBundle before processing any elements and FinishBundle before starting a new bundle.","Discard and rebuild plans whose status became Broken after a failed bundle.","When resuming split bundles, re-run StartBundle on the reconstructed plan before ProcessElement."],"exampleFix":"// before\nplan.Execute(ctx, id, mgr) // failed, plan now Broken\nplan.Execute(ctx, id2, mgr) // ProcessElement -> invalid status\n// after\nif plan.getStatus() == exec.Broken {\n    plan, err = constructAndExecutePlanWithContext(ctx, ...)\n}","handlingStrategy":"validation","validationCode":"if pardo.status != exec.Active {\n    return fmt.Errorf(\"pardo %v has no active bundle (status %v)\", pardo.UID, pardo.status)\n}\nerr := pardo.ProcessElement(ctx, elm, values...)","typeGuard":null,"tryCatchPattern":"err := plan.Execute(ctx, id, mgr)\nif err != nil {\n    // plan is Broken; do not attempt more elements\n    plan = nil\n    return err\n}","preventionTips":["Always pair StartBundle/FinishBundle around element processing.","Stop feeding elements as soon as Execute returns an error.","For split-bundle resume, ensure the resumed plan goes through StartBundle first."],"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"}