{"record":{"id":"42ac93cd1c83cb6b","repo":"apache/beam","slug":"ontimer-and-processelement-functions-for-dofn-should-have","errorCode":null,"errorMessage":"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","messagePattern":"OnTimer and ProcessElement functions for DoFn should have exactly same emitters, no\\. of emitters used in OnTimer: (.+?), no\\. of emitters used in ProcessElement: (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1410,"sourceCode":"\tif _, ok := fn.methods[onTimerName]; !ok {\n\t\terr := errors.Errorf(\"OnTimer function not defined for DoFn: %v\", fn.Name())\n\t\treturn errors.SetTopLevelMsgf(err, \"OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.\", fn.Name())\n\t}\n\n\tpipelineTimers, _ := fn.PipelineTimers()\n\n\tif _, ok := fn.methods[onTimerName].TimerProvider(); !ok {\n\t\terr := errors.Errorf(\"OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v\", fn.Name(), pipelineTimers)\n\t\treturn errors.SetTopLevelMsgf(err, \"OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v\"+\n\t\t\t\", Ensure that you are using the TimerProvider to set and clear the timers.\", fn.Name(), pipelineTimers)\n\t}\n\n\t_, otNum, otExists := fn.methods[onTimerName].Emits()\n\t_, peNum, peExists := fn.methods[processElementName].Emits()\n\n\tif otExists == peExists {\n\t\tif otNum != peNum {\n\t\t\treturn 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)\n\t\t}\n\t} else {\n\t\treturn 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)\n\t}\n\n\treturn nil\n}\n\nfunc validateTimer(fn *DoFn, numIn mainInputs) error {\n\tpt, fieldNames := fn.PipelineTimers()\n\n\tif _, ok := fn.methods[processElementName].TimerProvider(); ok {\n\t\tif numIn == MainSingle {\n\t\t\terr := errors.Errorf(\"ProcessElement uses a TimerProvider, but is not keyed\")\n\t\t\treturn errors.SetTopLevelMsgf(err, \"ProcessElement uses a TimerProvider, but is not keyed. \"+\n\t\t\t\t\"All stateful DoFns must take a key/value pair as an input.\")\n\t\t}\n\t\tif len(pt) == 0 {","sourceCodeStart":1392,"sourceCodeEnd":1428,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1392-L1428","documentation":"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.","triggerScenarios":"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).","commonSituations":"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).","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."],"exampleFix":"// before\nfunc (f *myFn) ProcessElement(ctx context.Context, x string, emitA func(string), emitB func(string))\nfunc (f *myFn) OnTimer(ctx context.Context, t timer.Event, emitA func(string))\n// after\nfunc (f *myFn) ProcessElement(ctx context.Context, x string, emitA func(string), emitB func(string))\nfunc (f *myFn) OnTimer(ctx context.Context, t timer.Event, emitA func(string), emitB func(string))","handlingStrategy":"validation","validationCode":"// Count emit parameters of both methods before registering the DoFn:\npeEmits := reflect.TypeOf(fn.ProcessElement) // compare with OnTimer signature at init/test time","typeGuard":"func emitterCountsMatch(dofn interface{}) bool { pe := reflect.TypeOf(dofn).Method(0); ot := reflect.TypeOf(dofn).Method(1); return countEmits(pe) == countEmits(ot) }","tryCatchPattern":"if err := beam.ParDo(scope, fn, input); err != nil { return fmt.Errorf(\"DoFn validation: %w\", err) }","preventionTips":["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"],"tags":["go","apache-beam","dofn","validation","timers"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}