apache/beam · error

prism error building stage

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.

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(...))
Defensive patterns

Strategy: validation

Validate before calling

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

Try / catch

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

Prevention

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

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

Appendix: 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)