apache/beam · error

unable to unmarshal TestStreamPayload for

Error message

unable to unmarshal TestStreamPayload for %v - %q: %w

What it means

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.

Solutions

1. Check SDK client and prism runner are built from compatible Apache Beam versions (TestStreamPayload proto must match).
2. Verify the pipeline was fully staged/serialized — resubmit the job rather than reusing a stale staging session.
3. If writing the pipeline programmatically, confirm the TestStream spec payload is populated before submission.
4. Reproduce with a minimal TestStream pipeline and file an issue with the wrapped cause (%w) message.

Example fix

// before: reusing a stale pipeline handle
job := stalePreparedPipeline // spec payload possibly truncated
runner.Submit(job)

// after: rebuild and resubmit a freshly serialized pipeline
pipeline := beam.NewPipeline()
teststream.New(pipeline, ...)
runner.Submit(serialize(pipeline)) // fresh, complete TestStreamPayload bytes

// Go: sanity-check the TestStream spec payload before submission
if len(spec.GetPayload()) == 0 {
    return fmt.Errorf("TestStream %q has empty spec payload", t.GetUniqueName())
}
var p pipepb.TestStreamPayload
if err := proto.Unmarshal(spec.GetPayload(), &p); err != nil {
    return fmt.Errorf("TestStream %q payload invalid: %w", t.GetUniqueName(), err)
}

err := runner.Prepare(req)
var re *runner.Err
if errors.As(err, &re) && strings.Contains(err.Error(), "unable to unmarshal TestStreamPayload") {
    // rebuild/reserialize the pipeline and resubmit
}

When it happens

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

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

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

• Failed to decode TestStreamPayload:
• failed to decode userfn
• prism error building stage
• unable to decode ParDoPayload for transform
• bad element

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go:235

					if early := trig.AfterEndOfWindow.GetEarlyFirings(); early == nil || early.GetNever() != nil {
						if ws.GetAllowedLateness() == 0 {
							// Late configuration doesn't matter, and there are no early firings.
							continue
						}
						if late := trig.AfterEndOfWindow.GetLateFirings(); late == nil || late.GetNever() != nil {
							// Lateness allowed, but but no firings anyway.
							continue
						}
					}
				}

				check("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))
			}

		case urns.TransformTestStream:
			var testStream pipepb.TestStreamPayload
			if err := proto.Unmarshal(t.GetSpec().GetPayload(), &testStream); err != nil {
				wrapped := fmt.Errorf("unable to unmarshal TestStreamPayload for %v - %q: %w", tid, t.GetUniqueName(), err)
				job.Failed(wrapped)
				return nil, wrapped
			}

			t.EnvironmentId = "" // Unset the environment, to ensure it's handled prism side.
			testStreamIds = append(testStreamIds, tid)

		default:
			// Composites can often have some unknown urn, permit those.
			// Eg. The Python SDK has urns "beam:transform:generic_composite:v1", "beam:transform:pickled_python:v1",
			// as well as the deprecated "beam:transform:read:v1", but they are composites.
			// We don't do anything special with these high level composites, but
			// we may be dealing with their internal subgraph already, so we ignore this transform.
			if len(t.GetSubtransforms()) > 0 {
				continue
			}
			// This may be an "empty" composite without subtransforms or a payload.
			// These just do PCollection manipulation which is already represented in the Pipeline graph.

View on GitHub (pinned to 12126d8942)