unable to encode ParDoPayload for

Error message

unable to encode ParDoPayload for %v in stage %v after rewrite

What it means

After prism rewrites timer coder IDs inside a ParDo payload, it re-marshals the ParDoPayload protobuf and writes it back to the transform spec. Failure of proto.Marshal means the in-memory payload could not be serialized, which is an internal inconsistency since the payload was originally valid protobuf.

Solutions

Inspect the wrapped error details and the affected transform ID for anomalous fields
Re-run the pipeline unchanged to rule out a transient protobuf marshal issue
File a Beam issue with the transform ID if reproducible; this indicates an internal invariant violation
Verify pipeline protobuf was produced by a supported SDK version

Defensive patterns

Strategy: retry

Try / catch

if err := job.Submit(ctx); err != nil {
    if strings.Contains(err.Error(), "unable to encode ParDoPayload") {
        return fmt.Errorf("payload marshal failure, retry or report: %w", err)
    }
    return err
}

Prevention

Avoid hand-editing pipeline protos
Keep protobuf library versions current
Report reproducible marshal failures to Beam

When it happens

Trigger: proto.MarshalOptions{}.Marshal(pardo) fails immediately after lpUnknownCoders mutated the ParDo's timer coder fields during buildDescriptor.

Common situations: Corrupted or hand-modified pipeline components; extremely large payloads hitting marshal limits; a prism/protobuf library bug rather than user error.

Understand the failure class

Background: json.Marshal / "failed to marshal" errors in Go: why "unsupported type" happens and how to fix it — this error's family across 22 libraries.

Related errors

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

Appendix: source

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

		for timerID, v := range pardo.GetTimerFamilySpecs() {
			stg.hasTimers = append(stg.hasTimers, engine.StaticTimerID{TransformID: tid, TimerFamily: timerID})
			if v.TimeDomain == pipepb.TimeDomain_PROCESSING_TIME {
				if stg.processingTimeTimers == nil {
					stg.processingTimeTimers = map[string]bool{}
				}
				stg.processingTimeTimers[timerID] = true
			}
			rewrite = true
			newCid, err := lpUnknownCoders(v.GetTimerFamilyCoderId(), coders, comps.GetCoders())
			if err != nil {
				return fmt.Errorf("unable to rewrite coder %v for timer %v for transform %v in stage %v: %w", v.GetTimerFamilyCoderId(), timerID, tid, stg.ID, err)
			}
			v.TimerFamilyCoderId = newCid
		}
		if rewrite {
			pyld, err := proto.MarshalOptions{}.Marshal(pardo)
			if err != nil {
				return fmt.Errorf("unable to encode ParDoPayload for %v in stage %v after rewrite", tid, stg.ID)
			}
			t.Spec.Payload = pyld
		}
	}
	if len(transforms) == 0 {
		return fmt.Errorf("buildDescriptor: invalid stage - no transforms at all %v", stg.ID)
	}

	// Start with outputs, since they're simple and uniform.
	sink2Col := map[string]string{}
	col2Coders := map[string]engine.PColInfo{}
	for _, o := range stg.outputs {
		col := clonePColToBundle(o.Global)
		wOutCid, err := makeWindowedValueCoder(o.Global, comps, coders)
		if err != nil {
			return fmt.Errorf("buildDescriptor: failed to handle coder on stage %v for output %+v, pcol %q %v:\n%w %v", stg.ID, o, o.Global, prototext.Format(col), err, stg.transforms)
		}
		sinkID := o.Transform + "_" + o.Local

View on GitHub (pinned to 12126d8942)