{"record":{"id":"f15fdcf02190428b","repo":"apache/beam","slug":"failed-to-decode-teststreampayload","errorCode":null,"errorMessage":"Failed to decode TestStreamPayload: ","messagePattern":"Failed to decode TestStreamPayload: ","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/handlerunner.go","lineNumber":226,"sourceCode":"\t\t\t}\n\t\t}\n\t}\n\n\t// Also recursively remove all sub-transforms.\n\ttoRemove = append(toRemove, removeSubTransforms(comps, t.GetSubtransforms())...)\n\n\t// Return the new components which is the transforms consumer\n\treturn prepareResult{\n\t\tSubbedComps:   nil, // Replace the reshuffle with nothing.\n\t\tRemovedLeaves: toRemove,\n\t\tForcedRoots:   forcedRoots,\n\t}\n}\n\nfunc (h *runner) handleTestStream(tid string, t *pipepb.PTransform, comps *pipepb.Components) prepareResult {\n\tvar pyld pipepb.TestStreamPayload\n\tif err := proto.Unmarshal(t.GetSpec().GetPayload(), &pyld); err != nil {\n\t\tpanic(\"Failed to decode TestStreamPayload: \" + err.Error())\n\t}\n\tcoders := map[string]*pipepb.Coder{}\n\t// Ensure awareness of the coder used for the teststream.\n\tocID := pyld.GetCoderId()\n\tcID, err := lpUnknownCoders(ocID, coders, comps.GetCoders())\n\tif err != nil {\n\t\tpanic(err)\n\t}\n\n\t// If the TestStream coder needs to be LP'ed or if it is a coder that has different\n\t// behaviors between nested context and outer context (in Java SDK), then we must\n\t// LP this coder and the TestStream data elements.\n\tforceLP := (cID != ocID && coders[ocID].GetSpec().GetUrn() != \"beam:go:coder:custom:v1\") ||\n\t\tcoders[ocID].GetSpec().GetUrn() == urns.CoderStringUTF8 ||\n\t\tcoders[ocID].GetSpec().GetUrn() == urns.CoderBytes ||\n\t\tcoders[ocID].GetSpec().GetUrn() == urns.CoderKV\n\n\tif !forceLP {","sourceCodeStart":208,"sourceCodeEnd":244,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/handlerunner.go#L208-L244","documentation":"handleTestStream unmarshals the transform's spec payload into pipepb.TestStreamPayload. If the bytes don't decode, the runner panics since TestStream handling entirely depends on the decoded events and coder ID.","triggerScenarios":"A TestStream transform is submitted whose spec payload is empty, truncated, or not a valid TestStreamPayload proto — e.g. from an SDK/test-harness version mismatch or a manually built TestStream transform.","commonSituations":"Running Beam integration tests against prism where SDK and runner protobuf definitions diverge; constructing TestStream pipelines in one SDK and executing with an incompatible prism build.","solutions":["Use the same Beam version for the SDK generating the TestStream and the prism runner executing it.","Verify the TestStream transform's payload decodes (e.g. with protoc) and contains events and a coder ID.","Rebuild the pipeline via the standard teststream package APIs so the payload is serialized by the SDK.","Report upstream with the pipeline dump if versions already match."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"pyld := &pipepb.TestStreamPayload{}\nif err := proto.Unmarshal(t.GetSpec().GetPayload(), pyld); err != nil { /* handle before running prism */ }","typeGuard":"func hasTestStreamPayload(t *pipepb.PTransform) bool { return len(t.GetSpec().GetPayload()) > 0 && t.GetSpec().GetUrn() == urns.TransformTestStream }","tryCatchPattern":"defer func(){ if r := recover(); r != nil && strings.Contains(fmt.Sprint(r), \"Failed to decode TestStreamPayload\") { /* handle decode failure */ } }()","preventionTips":["Use the teststream package APIs to build TestStreams","Align SDK and prism versions"],"tags":["go","protobuf","apache-beam","teststream"],"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"}