apache/beam · error

unable to decode ParDoPayload for

Error message

unable to decode ParDoPayload for %v in stage %v

What it means

buildDescriptor decodes every ParDo transform's payload when preparing a stage's bundle descriptor; failure to unmarshal a ParDoPayload returns this error identifying the transform and stage. Prism needs it to rewrite coder IDs so all coders are available in the bundle descriptor.

Solutions

1. Align the Beam SDK version with the prism runner version
2. Inspect the specific transform's payload bytes for validity
3. Rebuild and resubmit the pipeline; avoid reusing stale serialized pipelines

pardo := &pipepb.ParDoPayload{}
if err := (proto.UnmarshalOptions{}).Unmarshal(t.GetSpec().GetPayload(), pardo); err != nil {
    return fmt.Errorf("ParDo %v payload invalid: %w", tid, err)
}

if err := buildDescriptor(...); err != nil {
    if strings.Contains(err.Error(), "unable to decode ParDoPayload") {
        // dump the offending transform payload for inspection
    }
    return err
}

When it happens

Trigger: A transform with URN TransformParDo whose GetSpec().GetPayload() cannot be unmarshaled to pipepb.ParDoPayload during buildDescriptor.

Common situations: SDK/runner version mismatch changing payload contents; corrupted or hand-built pipeline protos; cross-language pipelines emitting unexpected payloads.

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

• bad type: , want
• base64 decode for failed
• computeFacts: unable to check
• invalid varint
• prism error building stage

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/stage.go:447

	clonePColToBundle := func(pid string) *pipepb.PCollection {
		col := proto.Clone(comps.GetPcollections()[pid]).(*pipepb.PCollection)
		pcollections[pid] = col
		return col
	}

	// Update coders for Stateful transforms.
	for _, tid := range stg.transforms {
		t := comps.GetTransforms()[tid]

		transforms[tid] = t

		if t.GetSpec().GetUrn() != urns.TransformParDo {
			continue
		}

		pardo := &pipepb.ParDoPayload{}
		if err := (proto.UnmarshalOptions{}).Unmarshal(t.GetSpec().GetPayload(), pardo); err != nil {
			return fmt.Errorf("unable to decode ParDoPayload for %v in stage %v", tid, stg.ID)
		}

		// We need to ensure the coders can be handled by prism, and are available in the bundle descriptor.
		// So we rewrite the transform's Payload with updated coder ids here.
		var rewrite bool
		var rewriteErr error
		for stateID, s := range pardo.GetStateSpecs() {
			rewrite = true
			rewriteCoder := func(cid *string) {
				newCid, err := lpUnknownCoders(*cid, coders, comps.GetCoders())
				if err != nil {
					rewriteErr = fmt.Errorf("unable to rewrite coder %v for state %v for transform %v in stage %v:%w", *cid, stateID, tid, stg.ID, err)
					return
				}
				*cid = newCid
			}
			switch s := s.GetSpec().(type) {
			case *pipepb.StateSpec_BagSpec:

View on GitHub (pinned to 12126d8942)