MergeAccumulators must be defined on accumulator %v

Error message

MergeAccumulators must be defined on accumulator %v

What it means

This panic in Combining.Add is a last-resort invariant check: a combiner accumulator was asked to merge, but no MergeAccumulators function is defined for the accumulator type. Earlier validation should have rejected such a Combining definition, so reaching this line means an invalid combine specification slipped through. The SDK panics because continuing would silently produce wrong aggregation results.

Source

Thrown at sdks/go/pkg/beam/core/state/state.go:322

		})
	}
	// If AddInput isn't defined, that means we must just have one accumulator type identical to the input type.
	if ma := p.MergeAccumulatorsFn(s.Key); ma != nil {
		var newVal any
		if f, ok := ma.(reflectx.Func2x1); ok {
			newVal = f.Call2x1(acc, val)
		} else {
			newVal = f.Call([]any{acc, val})[0]
		}
		return p.WriteValueState(Transaction{
			Key:  s.Key,
			Type: TransactionTypeSet,
			Val:  newVal,
		})
	}

	// Should be taken care of by previous validation
	panic(fmt.Sprintf("MergeAccumulators must be defined on accumulator %v", s))
}

// Read is used to read this instance of global pipeline state representing a combiner.
// When a value is not found, returns an empty list and false.
func (s *Combining[T1, T2, T3]) Read(p Provider) (T3, bool, error) {
	acc, ok, err := s.readAccumulator(p)
	if !ok || err != nil {
		var val T3
		return val, ok, err
	}

	if eo := p.ExtractOutputFn(s.Key); eo != nil {
		f, ok := eo.(reflectx.Func1x1)
		if ok {
			return f.Call1x1(acc).(T3), true, nil
		}
		return f.Call([]any{acc})[0].(T3), true, nil
	}

View on GitHub (pinned to 12126d8942)

Solutions

Define a MergeAccumulators method on the accumulator/DoFn type used in the Combining spec.
Prefer combine.CombinePerKey / combine.CombineGlobally helpers, which validate the signature at graph construction time.
Re-run signature validation on the DoFn (beam.Validate) before building the pipeline so the problem is caught as an error instead of a panic.

Example fix

// before
type sumFn struct{}
func (fn *sumFn) CreateAccumulator() int { return 0 }
func (fn *sumFn) AddInput(a, v int) int { return a + v }
// after
type sumFn struct{}
func (fn *sumFn) CreateAccumulator() int { return 0 }
func (fn *sumFn) AddInput(a, v int) int { return a + v }
func (fn *sumFn) MergeAccumulators(a, b int) int { return a + b }

Defensive patterns

Strategy: validation

Validate before calling

var fn interface{} = myCombinerFn{}
if _, ok := fn.(interface{ MergeAccumulators(int, int) int }); !ok { return errors.New("combiner missing MergeAccumulators") }

Type guard

func hasMergeAccumulators(fn interface{}) bool {
	_, ok := fn.(interface{ MergeAccumulators(interface{}, interface{}) interface{} })
	return ok
}

Prevention

Always define CreateAccumulator, AddInput, and MergeAccumulators together on combine DoFns.
Run beam.Validate on DoFns before pipeline construction.
Use combine.CombineGlobally / CombinePerKey helpers that validate signatures up front.

When it happens

Trigger: Constructing beam.Combining / state.Combining with a type that lacks a MergeAccumulators method in its DoFn signature, then calling Add on the combiner state; typically from a malformed combine transformation spec.

Common situations: Hand-writing combine function signatures instead of using combine.PerKey/CombinePerKey helpers; refactoring a DoFn and removing MergeAccumulators while the combiner setup still references it; using a custom accumulator type without the required merge method.

Related errors

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