{"record":{"id":"9931e7f67a08c496","repo":"apache/beam","slug":"unable-to-unmarshal-teststreampayload-for-v-q-w","errorCode":null,"errorMessage":"unable to unmarshal TestStreamPayload for %v - %q: %w","messagePattern":"unable to unmarshal TestStreamPayload for (.+?) - %q: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go","lineNumber":235,"sourceCode":"\t\t\t\t\tif early := trig.AfterEndOfWindow.GetEarlyFirings(); early == nil || early.GetNever() != nil {\n\t\t\t\t\t\tif ws.GetAllowedLateness() == 0 {\n\t\t\t\t\t\t\t// Late configuration doesn't matter, and there are no early firings.\n\t\t\t\t\t\t\tcontinue\n\t\t\t\t\t\t}\n\t\t\t\t\t\tif late := trig.AfterEndOfWindow.GetLateFirings(); late == nil || late.GetNever() != nil {\n\t\t\t\t\t\t\t// Lateness allowed, but but no firings anyway.\n\t\t\t\t\t\t\tcontinue\n\t\t\t\t\t\t}\n\t\t\t\t\t}\n\t\t\t\t}\n\n\t\t\t\tcheck(\"Unbounded GlobalWindow Triggered SideInput, are not currently supported by Prism. Sideinputs are only ready at end of window+allowed lateness. See https://github.com/apache/beam/issues/31438 for information.\", prototext.Format(ws))\n\t\t\t}\n\n\t\tcase urns.TransformTestStream:\n\t\t\tvar testStream pipepb.TestStreamPayload\n\t\t\tif err := proto.Unmarshal(t.GetSpec().GetPayload(), &testStream); err != nil {\n\t\t\t\twrapped := fmt.Errorf(\"unable to unmarshal TestStreamPayload for %v - %q: %w\", tid, t.GetUniqueName(), err)\n\t\t\t\tjob.Failed(wrapped)\n\t\t\t\treturn nil, wrapped\n\t\t\t}\n\n\t\t\tt.EnvironmentId = \"\" // Unset the environment, to ensure it's handled prism side.\n\t\t\ttestStreamIds = append(testStreamIds, tid)\n\n\t\tdefault:\n\t\t\t// Composites can often have some unknown urn, permit those.\n\t\t\t// Eg. The Python SDK has urns \"beam:transform:generic_composite:v1\", \"beam:transform:pickled_python:v1\",\n\t\t\t// as well as the deprecated \"beam:transform:read:v1\", but they are composites.\n\t\t\t// We don't do anything special with these high level composites, but\n\t\t\t// we may be dealing with their internal subgraph already, so we ignore this transform.\n\t\t\tif len(t.GetSubtransforms()) > 0 {\n\t\t\t\tcontinue\n\t\t\t}\n\t\t\t// This may be an \"empty\" composite without subtransforms or a payload.\n\t\t\t// These just do PCollection manipulation which is already represented in the Pipeline graph.","sourceCodeStart":217,"sourceCodeEnd":253,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go#L217-L253","documentation":"Prism's job preparation (Server.Prepare) failed to proto-unmarshal a TestStream transform's spec payload into a TestStreamPayload message. This means the serialized TestStream bytes sent by the pipeline are corrupt, empty, or not a valid TestStreamPayload. Prism aborts preparation and marks the job Failed.","triggerScenarios":"A pipeline containing a TestStream transform (urn TransformTestStream) is submitted to the prism runner via Prepare, and proto.Unmarshal of t.GetSpec().GetPayload() fails — e.g. the payload bytes are empty, truncated, or produced by an incompatible SDK version.","commonSituations":"Running beam testing pipeline fragments with TestStream against prism after an SDK/proto version mismatch; hand-crafted or corrupted pipeline protos; staging/session bugs that deliver an empty spec payload.","solutions":["Check SDK client and prism runner are built from compatible Apache Beam versions (TestStreamPayload proto must match).","Verify the pipeline was fully staged/serialized — resubmit the job rather than reusing a stale staging session.","If writing the pipeline programmatically, confirm the TestStream spec payload is populated before submission.","Reproduce with a minimal TestStream pipeline and file an issue with the wrapped cause (%w) message."],"exampleFix":"// before: reusing a stale pipeline handle\njob := stalePreparedPipeline // spec payload possibly truncated\nrunner.Submit(job)\n\n// after: rebuild and resubmit a freshly serialized pipeline\npipeline := beam.NewPipeline()\nteststream.New(pipeline, ...)\nrunner.Submit(serialize(pipeline)) // fresh, complete TestStreamPayload bytes","handlingStrategy":"validation","validationCode":"// Go: sanity-check the TestStream spec payload before submission\nif len(spec.GetPayload()) == 0 {\n    return fmt.Errorf(\"TestStream %q has empty spec payload\", t.GetUniqueName())\n}\nvar p pipepb.TestStreamPayload\nif err := proto.Unmarshal(spec.GetPayload(), &p); err != nil {\n    return fmt.Errorf(\"TestStream %q payload invalid: %w\", t.GetUniqueName(), err)\n}","typeGuard":null,"tryCatchPattern":"err := runner.Prepare(req)\nvar re *runner.Err\nif errors.As(err, &re) && strings.Contains(err.Error(), \"unable to unmarshal TestStreamPayload\") {\n    // rebuild/reserialize the pipeline and resubmit\n}","preventionTips":["Keep the SDK and prism runner on the same Beam version so protos match.","Always submit freshly serialized pipelines, not cached/stale handles.","Run TestStream pipelines against a local runner first to validate serialization."],"tags":["go","protobuf","beam-prism","teststream","deserialization"],"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-20T03:17:13.778Z"}