apache/beam · error

unable to decode ParDoPayload

Error message

unable to decode ParDoPayload

What it means

getSideInputs extracts the SideInputs map from a transform's ParDoPayload; this error is returned if the payload fails proto unmarshaling. Callers computeFacts and handleSideInput rely on it to know which inputs are side inputs and how to read them.

Source

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

		}
		slog.Debug("finalized bundle", "bundle", rb)
	}
	b.OutputData = engine.TentativeData{} // Clear the data.
	return nil
}

func getSideInputs(t *pipepb.PTransform) (map[string]*pipepb.SideInput, error) {
	switch t.GetSpec().GetUrn() {
	case urns.TransformParDo, urns.TransformProcessSizedElements, urns.TransformPairWithRestriction, urns.TransformSplitAndSize, urns.TransformTruncate:
		// Intentionally empty since these are permitted to have side inputs.
	default:
		// Nothing else is allowed to have side inputs.
		return nil, nil
	}
	// TODO, memoize this, so we don't need to repeatedly unmarshal.
	pardo := &pipepb.ParDoPayload{}
	if err := (proto.UnmarshalOptions{}).Unmarshal(t.GetSpec().GetPayload(), pardo); err != nil {
		return nil, fmt.Errorf("unable to decode ParDoPayload")
	}
	return pardo.GetSideInputs(), nil
}

func portFor(wInCid string, wk *worker.W) []byte {
	sourcePort := &fnpb.RemoteGrpcPort{
		CoderId: wInCid,
		ApiServiceDescriptor: &pipepb.ApiServiceDescriptor{
			Url: wk.Endpoint(),
		},
	}
	sourcePortBytes, err := proto.Marshal(sourcePort)
	if err != nil {
		slog.Error("bad port", slog.Any("error", err), slog.String("endpoint", sourcePort.ApiServiceDescriptor.GetUrl()))
	}
	return sourcePortBytes
}

View on GitHub (pinned to 12126d8942)

Solutions

1. Match SDK and prism Beam versions
2. Verify the transform URN corresponds to a ParDoPayload payload
3. Dump and decode the payload bytes locally to confirm corruption

Example fix

// cause wrapping for debuggability
if err := (proto.UnmarshalOptions{}).Unmarshal(t.GetSpec().GetPayload(), pardo); err != nil {
    return nil, fmt.Errorf("unable to decode ParDoPayload: %w", err)
}

sis, err := getSideInputs(t)
if err != nil {
    return fmt.Errorf("transform %v: %w", tID, err)
}

if sis, err := getSideInputs(t); err != nil {
    slog.Error("side input decode failed", "transform", tID, "err", err)
    return err
}

When it happens

Trigger: Calling getSideInputs on a transform whose spec payload is not decodable as a ParDoPayload — wrong payload for the URN, or truncated/corrupt bytes.

Common situations: Version skew between the SDK that produced the pipeline and prism; manually built pipeline protos attaching the wrong payload type to a ParDo URN.

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

• Iterators with timestamp values (<ET,V> and <ET, K, V>) are
• Type interface{} isn't a supported PCollection type
• failed to map main input window to side input window with Wi
• retrieveCoders: couldn't handle component %d %q of %q %v: %w
• prism error building stage %v - decoding TestStreamPayload:

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