{"record":{"id":"110cdb4a45c90e74","repo":"apache/beam","slug":"prism-error-building-stage-v-decoding-teststreampayload-w","errorCode":null,"errorMessage":"prism error building stage %v - decoding TestStreamPayload: \n%w","messagePattern":"prism error building stage (.+?) - decoding TestStreamPayload: \n%w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/execute.go","lineNumber":287,"sourceCode":"\t\t\t\t\tAllowedLateness: time.Duration(ws.GetAllowedLateness()) * time.Millisecond,\n\t\t\t\t\tAccumulating:    pipepb.AccumulationMode_ACCUMULATING == ws.GetAccumulationMode(),\n\t\t\t\t\tTrigger:         buildTrigger(ws.GetTrigger()),\n\t\t\t\t})\n\t\t\tcase urns.TransformImpulse:\n\t\t\t\timpulses = append(impulses, stage.ID)\n\t\t\t\tem.AddStage(stage.ID, nil, []string{getOnlyValue(t.GetOutputs())}, nil)\n\t\t\tcase urns.TransformTestStream:\n\t\t\t\t// Add a synthetic stage that should largely be unused.\n\t\t\t\tem.AddStage(stage.ID, nil, maps.Values(t.GetOutputs()), nil)\n\n\t\t\t\tfor pcolID, info := range stage.OutputsToCoders {\n\t\t\t\t\tem.RegisterPColInfo(pcolID, info)\n\t\t\t\t}\n\n\t\t\t\t// Decode the test stream, and convert it to the various events for the ElementManager.\n\t\t\t\tvar pyld pipepb.TestStreamPayload\n\t\t\t\tif err := proto.Unmarshal(t.GetSpec().GetPayload(), &pyld); err != nil {\n\t\t\t\t\treturn fmt.Errorf(\"prism error building stage %v - decoding TestStreamPayload: \\n%w\", stage.ID, err)\n\t\t\t\t}\n\n\t\t\t\ttsb := em.AddTestStream(stage.ID, t.Outputs)\n\t\t\t\tfor _, e := range pyld.GetEvents() {\n\t\t\t\t\tswitch ev := e.GetEvent().(type) {\n\t\t\t\t\tcase *pipepb.TestStreamPayload_Event_ElementEvent:\n\t\t\t\t\t\tvar elms []engine.TestStreamElement\n\t\t\t\t\t\tfor _, e := range ev.ElementEvent.GetElements() {\n\t\t\t\t\t\t\t// Encoded bytes are already handled in handleTestStream if needed.\n\t\t\t\t\t\t\telms = append(elms, engine.TestStreamElement{Encoded: e.GetEncodedElement(), EventTime: mtime.FromMilliseconds(e.GetTimestamp())})\n\t\t\t\t\t\t}\n\t\t\t\t\t\ttsb.AddElementEvent(ev.ElementEvent.GetTag(), elms)\n\t\t\t\t\tcase *pipepb.TestStreamPayload_Event_WatermarkEvent:\n\t\t\t\t\t\ttsb.AddWatermarkEvent(ev.WatermarkEvent.GetTag(), mtime.FromMilliseconds(ev.WatermarkEvent.GetNewWatermark()))\n\t\t\t\t\tcase *pipepb.TestStreamPayload_Event_ProcessingTimeEvent:\n\t\t\t\t\t\tif ev.ProcessingTimeEvent.GetAdvanceDuration() == int64(mtime.MaxTimestamp) {\n\t\t\t\t\t\t\t// TODO: Determine the SDK common formalism for setting processing time to infinity.\n\t\t\t\t\t\t\ttsb.AddProcessingTimeEvent(time.Duration(mtime.MaxTimestamp))","sourceCodeStart":269,"sourceCodeEnd":305,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/execute.go#L269-L305","documentation":"When building a stage backed by a TestStream transform, executePipeline unmarshals the transform's spec payload into pipepb.TestStreamPayload. If the proto bytes are invalid or not a TestStreamPayload, the stage build fails with this error including the stage ID and the unmarshal cause.","triggerScenarios":"proto.Unmarshal on a TestStream transform's spec payload fails — corrupted or truncated payload, wrong spec type at that transform ID, or cross-SDK encoding mismatches.","commonSituations":"Testing pipelines with TestStream where the submitting SDK emits a spec prism cannot parse (SDK/version skew), or hand-built pipelines with malformed TestStream specs.","solutions":["Verify the TestStream is constructed by the SDK's testing API, not manually serialized","Match SDK and prism/Beam versions so TestStreamPayload encodings agree","Inspect the wrapped error (%w) for the exact proto unmarshal failure","Check that the transform spec payload actually contains TestStreamPayload bytes, not another spec type"],"exampleFix":"// before: hand-rolled TestStream spec bytes\nspec := &pipepb.FunctionSpec{Urn: \"beam:transform:test:teststream:v1\", Payload: myBytes}\n// after: use the SDK test stream builder which emits a valid payload\nts := teststream.New(p, teststream.WithElements(...))","handlingStrategy":"validation","validationCode":"var pyld pipepb.TestStreamPayload\nif err := proto.Unmarshal(t.GetSpec().GetPayload(), &pyld); err != nil {\n  return fmt.Errorf(\"invalid TestStreamPayload before submit: %w\", err)\n}","typeGuard":null,"tryCatchPattern":"if strings.Contains(err.Error(), \"decoding TestStreamPayload\") {\n  // rebuild the test stream with the SDK's teststream builder\n}","preventionTips":["Always construct TestStream via the SDK testing API","Keep TestStream-producing SDK version aligned with prism"],"tags":["go","beam","prism-runner","teststream","protobuf"],"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"}