{"record":{"id":"34a690669d6ed544","repo":"apache/beam","slug":"unable-to-decode-pardopayload-for-v","errorCode":null,"errorMessage":"unable to decode ParDoPayload for %v","messagePattern":"unable to decode ParDoPayload for (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/preprocess.go","lineNumber":469,"sourceCode":"\tvar sideInputs []engine.LinkID\n\tinputs := map[string]bool{}\n\tfor pid, plinks := range stageFacts.PcolConsumers {\n\t\t// Check if this PCollection is generated in this bundle.\n\t\tif _, ok := stageFacts.PcolProducers[pid]; ok {\n\t\t\t// It is, so we will ignore for now.\n\t\t\tcontinue\n\t\t}\n\t\t// Add this collection to our input set.\n\t\tinputs[pid] = true\n\t\tfor _, link := range plinks {\n\t\t\tt := comps.GetTransforms()[link.Transform]\n\n\t\t\tvar sis map[string]*pipepb.SideInput\n\t\t\tswitch t.GetSpec().GetUrn() {\n\t\t\tcase urns.TransformParDo, urns.TransformProcessSizedElements, urns.TransformPairWithRestriction, urns.TransformSplitAndSize, urns.TransformTruncate:\n\t\t\t\tpardo := &pipepb.ParDoPayload{}\n\t\t\t\tif err := (proto.UnmarshalOptions{}).Unmarshal(t.GetSpec().GetPayload(), pardo); err != nil {\n\t\t\t\t\treturn fmt.Errorf(\"unable to decode ParDoPayload for %v\", link.Transform)\n\t\t\t\t}\n\t\t\t\tif pardo.GetRequestsFinalization() {\n\t\t\t\t\tstg.finalize = true\n\t\t\t\t}\n\t\t\t\tif len(pardo.GetTimerFamilySpecs())+len(pardo.GetStateSpecs())+len(pardo.GetOnWindowExpirationTimerFamilySpec()) > 0 {\n\t\t\t\t\tstg.stateful = true\n\t\t\t\t}\n\t\t\t\tif pardo.GetOnWindowExpirationTimerFamilySpec() != \"\" {\n\t\t\t\t\tstg.onWindowExpiration = engine.StaticTimerID{TransformID: link.Transform, TimerFamily: pardo.GetOnWindowExpirationTimerFamilySpec()}\n\t\t\t\t}\n\t\t\t\tsis = pardo.GetSideInputs()\n\t\t\t}\n\t\t\tif _, ok := sis[link.Local]; ok {\n\t\t\t\tsideInputs = append(sideInputs, engine.LinkID{Transform: link.Transform, Global: link.Global, Local: link.Local})\n\t\t\t} else {\n\t\t\t\tmainInputs[link.Global] = link.Global\n\t\t\t}\n\t\t}","sourceCodeStart":451,"sourceCodeEnd":487,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/preprocess.go#L451-L487","documentation":"finalizeStage inspects each ParDo-family transform in a stage and must decode its ParDoPayload; this error is returned when proto unmarshaling of the transform spec payload fails. Prism needs the payload to know about finalization requests, timers, and state specs.","triggerScenarios":"A transform with a ParDo-family URN (TransformParDo, ProcessSizedElements, PairWithRestriction, SplitAndSize, Truncate) whose GetSpec().GetPayload() bytes are not a valid ParDoPayload proto.","commonSituations":"Beam SDK/runner version skew where payload encoding changed; pipeline constructed by another tool that attached wrong payloads; corrupted serialized pipelines.","solutions":["Ensure SDK and prism runner are the same Beam version","Check that the transform URN actually matches the payload type actually stored","Re-generate/re-submit the pipeline rather than reusing a cached/corrupted job submission"],"exampleFix":"// verify payload before decode\nif len(t.GetSpec().GetPayload()) == 0 {\n    return fmt.Errorf(\"empty ParDoPayload for %v\", link.Transform)\n}","handlingStrategy":"validation","validationCode":"if len(t.GetSpec().GetPayload()) == 0 {\n    return fmt.Errorf(\"missing payload for ParDo %v\", link.Transform)\n}","typeGuard":null,"tryCatchPattern":"if err := finalizeStage(&stg, comps); err != nil {\n    slog.Error(\"stage finalize failed\", \"stage\", stg.ID, \"err\", err)\n    return err\n}","preventionTips":["Match SDK and runner Beam versions","Verify URN/payload pairing when building pipelines programmatically","Test pipelines end-to-end against prism before production"],"tags":["go","beam","protobuf","pardo"],"backgroundTag":"protobuf-unmarshal-failed","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"}