apache/beam · error

panic(err) propagating lpUnknownCoders error

Error message

panic(err) propagating lpUnknownCoders error

What it means

collectionPullDecoder resolves possibly unknown coders via lpUnknownCoders before building a pull decoder. If coder resolution fails (the coder ID chain is missing from the components), the error is propagated as a panic because the pipeline graph is deemed invalid.

Solutions

1. Upgrade Beam; unknown-coder handling improvements land regularly.
2. Inspect the submitted pipeline's coders map for the missing coder ID and ensure custom coders are registered via expansion or included in components.
3. Simplify or remove custom coder usage (e.g. use standard coders) to work around the resolution failure.
4. File a Beam issue with the job JSON if all coders appear valid.

// Validate coder chain resolves before submission
c := comps.GetCoders()[coldCId]
if c == nil {
    return fmt.Errorf("coder %s missing from components", coldCId)
}
for _, comp := range c.GetComponentCoders() {
    if comps.GetCoders()[comp] == nil {
        return fmt.Errorf("component coder %s missing", comp)
    }
}

cID, err := lpUnknownCoders(coldCId, coders, comps.GetCoders())
if err != nil {
    return nil, fmt.Errorf("resolving coder %s: %w", coldCId, err)
}

When it happens

Trigger: A PCollection references a coder ID that does not exist in the components' coder map, or lpUnknownCoders cannot reduce an unknown-leaf coder to a concrete one when building the stage's element decoder.

Common situations: Pipelines submitted with custom coders prism doesn't recognize, cross-language pipelines whose coder components weren't fully materialized, or hand-crafted/optimized job submissions with dropped components.

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.

Related errors

• expected single value map, had
• coder must not be nil
• computeFacts: two producers for one PCollection
• err
• err

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/execute.go:401

				if err := s.Execute(egctx, j, wk, comps, em, rb); err != nil {
					// Ensure we clean up on bundle failure
					j.Logger.Error("Bundle Failed.", slog.Any("error", err))
					em.FailBundle(rb)
					return err
				}
				return nil
			})
		// Log a heartbeat every 60 seconds
		case <-ticker.C:
			j.Logger.Info("pipeline is running", slog.String("job", j.String()))
		}
	}
}

func collectionPullDecoder(coldCId string, coders map[string]*pipepb.Coder, comps *pipepb.Components) func(io.Reader) []byte {
	cID, err := lpUnknownCoders(coldCId, coders, comps.GetCoders())
	if err != nil {
		panic(err)
	}
	return pullDecoder(coders[cID], coders)
}

func extractKVCoderID(coldCId string, coders map[string]*pipepb.Coder) (string, bool) {
	c := coders[coldCId]
	if c.GetSpec().GetUrn() == urns.CoderKV {
		return c.GetComponentCoderIds()[0], true
	}
	return "", false
}

func getWindowValueCoders(comps *pipepb.Components, col *pipepb.PCollection, coders map[string]*pipepb.Coder) (engine.WinCoderType, exec.WindowDecoder, exec.WindowEncoder) {
	ws := comps.GetWindowingStrategies()[col.GetWindowingStrategyId()]
	wcID, err := lpUnknownCoders(ws.GetWindowCoderId(), coders, comps.GetCoders())
	if err != nil {
		panic(err)
	}

View on GitHub (pinned to 12126d8942)