apache/beam · error

unable to decode ParDoPayload for transform

Error message

unable to decode ParDoPayload for transform[%v]

What it means

In handlecombine's PrepareTransform, the transform's spec payload is expected to be a CombinePayload protobuf. If proto.Unmarshal fails, prism panics with a message (misleadingly) naming ParDoPayload, since a Combine transform's spec must always decode as CombinePayload; failure means the job graph's spec bytes are corrupt or the transform is not actually a Combine.

Solutions

1. Check that the transform's URN actually matches a Combine; report a routing bug to Beam if a non-Combine payload reaches this handler.
2. Align SDK and prism (runner) versions so payloads serialize compatibly.
3. Regenerate/re-expand the pipeline rather than hand-editing job JSON.
4. Inspect the transform payload bytes in the job JSON to confirm corruption.

// Verify the payload decodes as CombinePayload before running the stage
cmb := &pipepb.CombinePayload{}
if err := (proto.UnmarshalOptions{}).Unmarshal(t.GetSpec().GetPayload(), cmb); err != nil {
    return fmt.Errorf("transform %s is not a valid Combine: %w", t.GetUniqueName(), err)
}

if err := recover(); err != nil { /* around PrepareTransform */ }

When it happens

Trigger: A PTransform routed to the Combine handler whose spec payload is not valid CombinePayload bytes — e.g. wrong URN routing, truncated payload from a malformed submission, or a runner/SDK version mismatch producing an incompatible payload format.

Common situations: Cross-language Combine expansion emitting unexpected payloads, submitting a job graph generated by a newer SDK than the prism runner supports, or hand-edited pipeline JSON.

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

• expected single value map, had
• invalid CombinePayload payload for
• panic(err) propagating lpUnknownCoders error
• prism consistency error: trying to remove a non-existent…
• prism consistency error: trying to remove a timer for a key…

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/handlecombine.go:156

	//
	// Then we can produce the PCollections.
	// We can reuse the INPUT and OUTPUT PCollections.
	// We need LIFTED to have KV<K, A>  kv_k_a
	// We need GROUPED_A to have KV<K, Iter<A>> kv_k_iter_a
	// We need MERGED_A to have KV<K, A> kv_k_a
	//
	// GROUPED_I ends up unused.
	//
	// The PCollections inherit the properties of the Input PCollection
	// such as Boundedness, and Windowing Strategy.
	//
	// With these, we can produce the PTransforms with the appropriate URNs for the
	// different parts of the composite, and return the new components.

	cmbPayload := t.GetSpec().GetPayload()
	cmb := &pipepb.CombinePayload{}
	if err := (proto.UnmarshalOptions{}).Unmarshal(cmbPayload, cmb); err != nil {
		panic(fmt.Sprintf("unable to decode ParDoPayload for transform[%v]", t.GetUniqueName()))
	}

	// First lets get the key coder ID.
	var pcolInID string
	// There's only one input.
	for _, pcol := range t.GetInputs() {
		pcolInID = pcol
	}
	inputPCol := comps.GetPcollections()[pcolInID]
	kvkiID := inputPCol.GetCoderId()
	kID := comps.GetCoders()[kvkiID].GetComponentCoderIds()[0]

	// Now we can start synthesis!
	// Coder IDs
	aID := cmb.AccumulatorCoderId

	ckvprefix := "c" + tid + "_kv_"

View on GitHub (pinned to 12126d8942)