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).

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) }
Defensive patterns

Strategy: validation

Validate before calling

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)
}

Type guard

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

Try / catch

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

Prevention

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

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

Appendix: 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)