apache/beam · error

panic(err) propagating lpUnknownCoders error

Error message

panic(err) propagating lpUnknownCoders error

What it means

After decoding the TestStreamPayload, handleTestStream calls lpUnknownCoders to bring the TestStream's coder into the runner's known-coder set. Any error returned is re-raised as a panic because without a resolvable coder the TestStream's elements cannot be decoded or planned.

Solutions

1. Check the TestStream's coder_id resolves to a coder present in the job's components.
2. Rebuild the pipeline through the SDK's teststream APIs so all referenced coders are registered in components.
3. Align SDK and prism versions so coder URNs used by the TestStream are understood by lpUnknownCoders.
4. Log/inspect the wrapped err for the specific missing coder ID and add or fix that coder definition.

if comps.GetCoders()[pyld.GetCoderId()] == nil { /* coder missing from components; fix graph */ }

defer func(){ if r := recover(); r != nil { if e, ok := r.(error); ok { /* log e (lpUnknownCoders failure) */ } } }()

When it happens

Trigger: The TestStream's coder ID refers to a coder missing from components.coders, or references nested/unknown coders that lpUnknownCoders fails to look up or convert.

Common situations: Cross-SDK TestStream pipelines where coder definitions weren't shipped in the job components; hand-rolled TestStream pipelines that reference a coder ID not registered in the pipeline graph.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• Expected single input PCollection in reshuffle:
• Expected single output PCollection in reshuffle:
• forceLpCoders: coder
• lpUnknownCoders: coder
• lpUnknownCoders: couldn't handle component

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

Appendix: source

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

	// 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 {
		return prepareResult{SubbedComps: &pipepb.Components{
			Transforms: map[string]*pipepb.PTransform{tid: t},
		}}
	}

	var mustLP func(v []byte) []byte
	if coders[ocID].GetSpec().GetUrn() != urns.CoderKV {

View on GitHub (pinned to 12126d8942)