OnTimer and ProcessElement functions for DoFn should have ex

Error message

OnTimer and ProcessElement functions for DoFn should have exactly same emitters, no. of emitters used in OnTimer: %v, no. of emitters used in ProcessElement: %v

What it means

During DoFn validation, Beam requires a user-defined OnTimer method's emitter count to exactly equal ProcessElement's emitter count when both methods exist. This error fires when both methods exist but use a different number of output emitters, so the framework cannot wire timer outputs consistently.

Source

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

	if _, ok := fn.methods[onTimerName]; !ok {
		err := errors.Errorf("OnTimer function not defined for DoFn: %v", fn.Name())
		return errors.SetTopLevelMsgf(err, "OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.", fn.Name())
	}

	pipelineTimers, _ := fn.PipelineTimers()

	if _, ok := fn.methods[onTimerName].TimerProvider(); !ok {
		err := errors.Errorf("OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v", fn.Name(), pipelineTimers)
		return errors.SetTopLevelMsgf(err, "OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v"+
			", Ensure that you are using the TimerProvider to set and clear the timers.", fn.Name(), pipelineTimers)
	}

	_, otNum, otExists := fn.methods[onTimerName].Emits()
	_, peNum, peExists := fn.methods[processElementName].Emits()

	if otExists == peExists {
		if otNum != peNum {
			return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, no. of emitters used in OnTimer: %v, no. of emitters used in ProcessElement: %v", otNum, peNum)
		}
	} else {
		return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, emitters used in OnTimer: %v, emitters used in ProcessElement: %v", otExists, peExists)
	}

	return nil
}

func validateTimer(fn *DoFn, numIn mainInputs) error {
	pt, fieldNames := fn.PipelineTimers()

	if _, ok := fn.methods[processElementName].TimerProvider(); ok {
		if numIn == MainSingle {
			err := errors.Errorf("ProcessElement uses a TimerProvider, but is not keyed")
			return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but is not keyed. "+
				"All stateful DoFns must take a key/value pair as an input.")
		}
		if len(pt) == 0 {

View on GitHub (pinned to 12126d8942)

Solutions

Make OnTimer emit to exactly the same number of emitters as ProcessElement.
If OnTimer does not need to emit to all outputs, emit zero-value or placeholder values on the remaining emitters.
Split logic so outputs only produced from ProcessElement live in a separate DoFn without timers.
Review the DoFn's emit signature declarations to keep them consistent across both methods.

Example fix

// before
func (f *myFn) ProcessElement(ctx context.Context, x string, emitA func(string), emitB func(string))
func (f *myFn) OnTimer(ctx context.Context, t timer.Event, emitA func(string))
// after
func (f *myFn) ProcessElement(ctx context.Context, x string, emitA func(string), emitB func(string))
func (f *myFn) OnTimer(ctx context.Context, t timer.Event, emitA func(string), emitB func(string))

Defensive patterns

Strategy: validation

Validate before calling

// Count emit parameters of both methods before registering the DoFn:
peEmits := reflect.TypeOf(fn.ProcessElement) // compare with OnTimer signature at init/test time

Type guard

func emitterCountsMatch(dofn interface{}) bool { pe := reflect.TypeOf(dofn).Method(0); ot := reflect.TypeOf(dofn).Method(1); return countEmits(pe) == countEmits(ot) }

Try / catch

if err := beam.ParDo(scope, fn, input); err != nil { return fmt.Errorf("DoFn validation: %w", err) }

Prevention

Mirror ProcessElement's emit signature in OnTimer whenever you add timers
Add a unit test constructing the DoFn to catch validation early
Change emit signatures on both methods in the same commit

When it happens

Trigger: Defining a DoFn where ProcessElement emits to N outputs but OnTimer emits to M outputs (N != M), then registering the DoFn in a pipeline (via beam.ParDo or similar).

Common situations: Adding an OnTimer method for state/timer support and forgetting to mirror the ProcessElement emit signature; emitting to an extra output only from ProcessElement (e.g. a dead-letter output).

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

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