apache/beam · error

failed to construct userfn

Error message

failed to construct userfn

What it means

After successfully decoding the userfn, decodeFn calls funcx.New(reflectx.MakeFunc(fn)) to validate and wrap the function's signature; failure means the reconstructed function has a signature funcx cannot analyze (not a valid func, unsupported parameter/return kinds).

Source

Thrown at sdks/go/pkg/beam/core/runtime/graphx/serialize.go:297

			wrapped := errors.Wrap(err, "bad type")
			return nil, errors.WithContextf(wrapped, "failed to decode dynamic DoFn %v", u)
		}
		return graph.NewFn(&graph.DynFn{
			Name: u.Dynfn.Name,
			T:    t,
			Data: u.Dynfn.Data,
			Gen:  gen.(func(string, reflect.Type, []byte) reflectx.Func),
		})
	}
	if u.Fn != nil {
		fn, err := decodeUserFn(u.Fn)
		if err != nil {
			wrapped := errors.Wrap(err, "failed to decode userfn")
			return nil, errors.WithContextf(wrapped, "decoding DoFn %v", u)
		}
		fx, err := funcx.New(reflectx.MakeFunc(fn))
		if err != nil {
			wrapped := errors.Wrap(err, "failed to construct userfn")
			return nil, errors.WithContextf(wrapped, "decoding DoFn %v", u)
		}
		return &graph.Fn{Fn: fx}, nil
	}

	t, err := decodeType(u.Type)
	if err != nil {
		wrapped := errors.Wrap(err, "bad type")
		return nil, errors.WithContextf(wrapped, "decoding structural DoFn %v", u)
	}
	elem := reflect.New(t)
	if err := jsonx.UnmarshalFrom(elem.Interface(), strings.NewReader(u.Opt)); err != nil {
		wrapped := errors.Wrap(err, "bad struct encoding")
		return nil, errors.WithContextf(wrapped, "decoding structural DoFn %v", u)
	}
	fn := elem.Elem().Interface()
	return graph.NewFn(fn)
}

View on GitHub (pinned to 12126d8942)

Solutions

1. Check the wrapped funcx error for the exact signature complaint and fix the DoFn's method signature.
2. Ensure the registered fn is actually a func with supported parameter/return types (context, iterators, emitters).
3. Use standard DoFn signatures matching current Beam Go docs.
4. Align Beam versions between producer and consumer of the pipeline proto.

Example fix

// before
func (f *myFn) Process() int { return 1 } // unsupported signature

// after
func (f *myFn) ProcessElement(ctx context.Context, elm int, emit func(int)) { emit(elm) }

if reflect.ValueOf(fn).Kind() != reflect.Func {
    return fmt.Errorf("userfn must be a func, got %T", fn)
}
if _, err := funcx.New(reflectx.MakeFunc(fn)); err != nil {
    return fmt.Errorf("unsupported fn signature: %w", err)
}

func isFunc(v interface{}) bool { return v != nil && reflect.ValueOf(v).Kind() == reflect.Func }

fx, err := funcx.New(reflectx.MakeFunc(fn))
if err != nil {
    return fmt.Errorf("DoFn signature not supported by funcx: %w", err)
}

When it happens

Trigger: DecodeMultiEdge decoding a Fn whose decoded userfn value is not a valid function or has a signature outside funcx's supported DoFn/emit shapes.

Common situations: DoFn registered as a non-func value; signature shapes from custom/dynamic fn generation that violate funcx expectations; version skew producing mismatched expectations.

Related errors

• may only have a single context.Context parameter and it must
• may only have a single PaneInfo parameter and it must preced
• may only have a single Window parameter and it must precede
• may only have a single beam.EventTime parameter and it must
• Incompatible func type: got func %v with %v inputs and %v ou

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