apache/beam · error

failed to add input kind

Error message

failed to add input kind: %v

What it means

Generic wrap used inside addMultiEdge's handleErr helper: any failure while adding an input kind (encoding a multi-edge's data, coders, windows, state specs, etc.) is re-wrapped with the edge's name. The inner error carries the real cause; this message identifies which edge of the graph failed.

Solutions

1. Print the full wrapped error chain to find the root cause
2. Locate the edge named in the message in your pipeline and inspect its inputs/coders
3. Reproduce with a minimal pipeline containing only that transform
4. Check SDK version known issues for graphx marshalling

if _, err := graphx.Marshal(p); err != nil {
    var e *errors.Error
    if stderrors.As(err, &e) && strings.Contains(err.Error(), "failed to add input kind") {
        log.Printf("edge failure, root cause: %v", errors.Unwrap(err))
    }
    return err
}

When it happens

Trigger: Any failure inside addMultiEdge while marshalling — e.g. coder registration failures, windowing strategy errors, external payload mismatches — all funneled through handleErr with the NamedEdge in the message.

Common situations: Debugging large pipeline graphs where the marshal error chain names the specific edge/transform that failed to convert.

Related errors

• failed to add window strategy
• AfterProcessingTime trigger set without a delay or…
• array len mismatch. decoding
• At least one subtrigger required for composite triggers.
• attempted to add namespace to missing coder id

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/graphx/translate.go:367

func getSideWindowMappingUrn(winFn *window.Fn) string {
	var mappingUrn string
	switch winFn.Kind {
	case window.GlobalWindows:
		mappingUrn = URNWindowMappingGlobal
	case window.FixedWindows:
		mappingUrn = URNWindowMappingFixed
	case window.SlidingWindows:
		mappingUrn = URNWindowMappingSliding
	case window.Sessions:
		panic("session windowing is not supported for side inputs")
	}
	return mappingUrn
}

func (m *marshaller) addMultiEdge(edge NamedEdge) ([]string, error) {
	handleErr := func(err error) ([]string, error) {
		return nil, errors.Wrapf(err, "failed to add input kind: %v", edge)
	}
	id := edgeID(edge.Edge)
	if _, exists := m.transforms[id]; exists {
		return []string{id}, nil
	}

	switch {
	case edge.Edge.Op == graph.CoGBK && len(edge.Edge.Input) > 1:
		cogbkID, err := m.expandCoGBK(edge)
		if err != nil {
			return handleErr(err)
		}
		return []string{cogbkID}, nil
	case edge.Edge.Op == graph.Reshuffle:
		reshuffleID, err := m.expandReshuffle(edge)
		if err != nil {
			return handleErr(err)
		}

View on GitHub (pinned to 12126d8942)