apache/beam · error

prism error building stage %v - decoding TestStreamPayload:

Error message

prism error building stage %v - decoding TestStreamPayload: 
%w

What it means

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.

Source

Thrown at sdks/go/pkg/beam/runners/prism/internal/execute.go:287

					AllowedLateness: time.Duration(ws.GetAllowedLateness()) * time.Millisecond,
					Accumulating:    pipepb.AccumulationMode_ACCUMULATING == ws.GetAccumulationMode(),
					Trigger:         buildTrigger(ws.GetTrigger()),
				})
			case urns.TransformImpulse:
				impulses = append(impulses, stage.ID)
				em.AddStage(stage.ID, nil, []string{getOnlyValue(t.GetOutputs())}, nil)
			case urns.TransformTestStream:
				// Add a synthetic stage that should largely be unused.
				em.AddStage(stage.ID, nil, maps.Values(t.GetOutputs()), nil)

				for pcolID, info := range stage.OutputsToCoders {
					em.RegisterPColInfo(pcolID, info)
				}

				// Decode the test stream, and convert it to the various events for the ElementManager.
				var pyld pipepb.TestStreamPayload
				if err := proto.Unmarshal(t.GetSpec().GetPayload(), &pyld); err != nil {
					return fmt.Errorf("prism error building stage %v - decoding TestStreamPayload: \n%w", stage.ID, err)
				}

				tsb := em.AddTestStream(stage.ID, t.Outputs)
				for _, e := range pyld.GetEvents() {
					switch ev := e.GetEvent().(type) {
					case *pipepb.TestStreamPayload_Event_ElementEvent:
						var elms []engine.TestStreamElement
						for _, e := range ev.ElementEvent.GetElements() {
							// Encoded bytes are already handled in handleTestStream if needed.
							elms = append(elms, engine.TestStreamElement{Encoded: e.GetEncodedElement(), EventTime: mtime.FromMilliseconds(e.GetTimestamp())})
						}
						tsb.AddElementEvent(ev.ElementEvent.GetTag(), elms)
					case *pipepb.TestStreamPayload_Event_WatermarkEvent:
						tsb.AddWatermarkEvent(ev.WatermarkEvent.GetTag(), mtime.FromMilliseconds(ev.WatermarkEvent.GetNewWatermark()))
					case *pipepb.TestStreamPayload_Event_ProcessingTimeEvent:
						if ev.ProcessingTimeEvent.GetAdvanceDuration() == int64(mtime.MaxTimestamp) {
							// TODO: Determine the SDK common formalism for setting processing time to infinity.
							tsb.AddProcessingTimeEvent(time.Duration(mtime.MaxTimestamp))

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the TestStream is constructed by the SDK's testing API, not manually serialized
2. Match SDK and prism/Beam versions so TestStreamPayload encodings agree
3. Inspect the wrapped error (%w) for the exact proto unmarshal failure
4. Check that the transform spec payload actually contains TestStreamPayload bytes, not another spec type

Example fix

// before: hand-rolled TestStream spec bytes
spec := &pipepb.FunctionSpec{Urn: "beam:transform:test:teststream:v1", Payload: myBytes}
// after: use the SDK test stream builder which emits a valid payload
ts := teststream.New(p, teststream.WithElements(...))

var pyld pipepb.TestStreamPayload
if err := proto.Unmarshal(t.GetSpec().GetPayload(), &pyld); err != nil {
  return fmt.Errorf("invalid TestStreamPayload before submit: %w", err)
}

if strings.Contains(err.Error(), "decoding TestStreamPayload") {
  // rebuild the test stream with the SDK's teststream builder
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• retrieveCoders: couldn't handle component %d %q of %q %v: %w
• prism error building stage %v - unknown TestStream event typ
• pipeline canceled
• makeWindowedValueCoder: couldn't process coder for pcollecti
• lpUnknownCoders: coder %q not present in base map

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/110cdb4a45c90e74. Report an issue: GitHub.