apache/beam · error

Failed to decode TestStreamPayload:

Error message

Failed to decode TestStreamPayload: 

What it means

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.

Solutions

1. Use the same Beam version for the SDK generating the TestStream and the prism runner executing it.
2. Verify the TestStream transform's payload decodes (e.g. with protoc) and contains events and a coder ID.
3. Rebuild the pipeline via the standard teststream package APIs so the payload is serialized by the SDK.
4. Report upstream with the pipeline dump if versions already match.

pyld := &pipepb.TestStreamPayload{}
if err := proto.Unmarshal(t.GetSpec().GetPayload(), pyld); err != nil { /* handle before running prism */ }

func hasTestStreamPayload(t *pipepb.PTransform) bool { return len(t.GetSpec().GetPayload()) > 0 && t.GetSpec().GetUrn() == urns.TransformTestStream }

defer func(){ if r := recover(); r != nil && strings.Contains(fmt.Sprint(r), "Failed to decode TestStreamPayload") { /* handle decode failure */ } }()

When it happens

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

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

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.

• Parsing and encoding errors: unexpected token, malformed input — why parsers reject input and how to find the real culprit.

Related errors

• empty type
• failed to decode userfn
• failed to marshal payload as proto
• invalid CombinePayload payload for
• invalid ParDo payload for

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/handlerunner.go:226

			}
		}
	}

	// Also recursively remove all sub-transforms.
	toRemove = append(toRemove, removeSubTransforms(comps, t.GetSubtransforms())...)

	// Return the new components which is the transforms consumer
	return prepareResult{
		SubbedComps:   nil, // Replace the reshuffle with nothing.
		RemovedLeaves: toRemove,
		ForcedRoots:   forcedRoots,
	}
}

func (h *runner) handleTestStream(tid string, t *pipepb.PTransform, comps *pipepb.Components) prepareResult {
	var pyld pipepb.TestStreamPayload
	if err := proto.Unmarshal(t.GetSpec().GetPayload(), &pyld); err != nil {
		panic("Failed to decode TestStreamPayload: " + err.Error())
	}
	coders := map[string]*pipepb.Coder{}
	// Ensure awareness of the coder used for the teststream.
	ocID := pyld.GetCoderId()
	cID, err := lpUnknownCoders(ocID, coders, comps.GetCoders())
	if err != nil {
		panic(err)
	}

	// If the TestStream coder needs to be LP'ed or if it is a coder that has different
	// behaviors between nested context and outer context (in Java SDK), then we must
	// LP this coder and the TestStream data elements.
	forceLP := (cID != ocID && coders[ocID].GetSpec().GetUrn() != "beam:go:coder:custom:v1") ||
		coders[ocID].GetSpec().GetUrn() == urns.CoderStringUTF8 ||
		coders[ocID].GetSpec().GetUrn() == urns.CoderBytes ||
		coders[ocID].GetSpec().GetUrn() == urns.CoderKV

	if !forceLP {

View on GitHub (pinned to 12126d8942)