apache/beam · error

invalid CombineFn

Error message

invalid CombineFn

What it means

NewCombineFn wraps NewFn and AsCombineFn; if the supplied value is not a valid CombineFn (wrong kind of function/struct or methods missing), the underlying error is wrapped with context 'constructing CombineFn' and surfaced as 'invalid CombineFn'. Beam only accepts structs implementing the CombineFn method set (CreateAccumulator/AddInput/MergeAccumulators/ExtractOutput) or MergeAccumulators-only functions.

Solutions

1. Inspect the wrapped inner error via errors.Context for the exact reason NewFn failed.
2. Ensure the value is either a struct with the CombineFn method set or a func with MergeAccumulators-compatible signature.
3. Export the CombineFn methods (MergeAccumulators, CreateAccumulator, AddInput, ExtractOutput).
4. Verify accumulator type is consistent across all CombineFn methods.

// verify at init that the combine value implements the required method set
var _ = func() error {
    if _, err := graph.NewCombineFn(MyCombine{}); err != nil { return err }
    return nil
}

func isCombineFn(v interface{}) bool {
    m := reflect.ValueOf(v)
    return m.MethodByName("MergeAccumulators").IsValid()
}

When it happens

Trigger: Calling beam.Combine/Add(beam.CombinePerKey, someValue) with a value that is neither a valid combine struct nor a function, or one whose CombineFn methods have wrong signatures.

Common situations: Passing a plain function that isn't a merge-accumulators-only signature; passing a struct that's missing MergeAccumulators; typos in method names (lowercase unexported methods); passing a value of a non-func type entirely.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• bad combinefn
• Duplicate timer family ID
• : failed to find required method on type
• : requires at least 1 return value.
• bad coder kind

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/graph/fn.go:1510

func (f *CombineFn) CompactFn() *funcx.Fn {
	return f.methods[compactName]
}

// TeardownFn returns the "Teardown" function, if present.
func (f *CombineFn) TeardownFn() *funcx.Fn {
	return f.methods[teardownName]
}

// Name returns the name of the function or struct.
func (f *CombineFn) Name() string {
	return (*Fn)(f).Name()
}

// NewCombineFn constructs a CombineFn from the given value, if possible.
func NewCombineFn(fn any) (*CombineFn, error) {
	ret, err := NewFn(fn)
	if err != nil {
		return nil, errors.WithContext(errors.Wrapf(err, "invalid CombineFn"), "constructing CombineFn")
	}
	return AsCombineFn(ret)
}

// AsCombineFn converts a Fn to a CombineFn, if possible.
func AsCombineFn(fn *Fn) (*CombineFn, error) {
	const fnKind = "graph.AsCombineFn"
	if fn.methods == nil {
		fn.methods = make(map[string]*funcx.Fn)
	}
	if fn.Fn != nil {
		fn.methods[mergeAccumulatorsName] = fn.Fn
	}

	mergeFn, ok := fn.methods[mergeAccumulatorsName]
	if !ok {
		return nil, errors.Errorf("%v: failed to find required %v method on type: %v", fnKind, mergeAccumulatorsName, fn.Name())
	}

View on GitHub (pinned to 12126d8942)