{"record":{"id":"e85f56c1cdd01cc1","repo":"apache/beam","slug":"ontimer-and-processelement-functions-for-dofn-should-have-fn","errorCode":null,"errorMessage":"OnTimer and ProcessElement functions for DoFn should have exactly same emitters, emitters used in OnTimer: %v, emitters used in ProcessElement: %v","messagePattern":"OnTimer and ProcessElement functions for DoFn should have exactly same emitters, emitters used in OnTimer: (.+?), emitters used in ProcessElement: (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1413,"sourceCode":"\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 {\n\t\t\terr := errors.New(\"ProcessElement uses a TimerProvider, but no Timer fields are defined in the DoFn\")\n\t\t\treturn errors.SetTopLevelMsgf(err, \"ProcessElement uses a TimerProvider, but no timer fields are defined in the DoFn\"+\n\t\t\t\t\", Ensure that your DoFn exports the Timer fields used to set and clear timers.\")","sourceCodeStart":1395,"sourceCodeEnd":1431,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1395-L1431","documentation":"Companion to the emitter-count mismatch: Beam requires that either both OnTimer and ProcessElement emit outputs or neither does. This error fires when one method has emitters and the other does not (the existence flags differ), which the framework treats as an invalid DoFn shape.","triggerScenarios":"Defining a DoFn whose ProcessElement has emit parameters but whose OnTimer has none (or vice versa), then validating/registering the DoFn in a pipeline.","commonSituations":"Adding OnTimer to an existing emitting DoFn with a bare context-only signature; removing all emit parameters from OnTimer during refactoring while ProcessElement still emits.","solutions":["Add emit parameters to OnTimer matching ProcessElement's outputs (or remove them from both).","If OnTimer legitimately produces no output, remove emit parameters from ProcessElement too, or move emitting logic into another DoFn.","Keep both method signatures in sync whenever you change output types."],"exampleFix":"// before\nfunc (f *myFn) ProcessElement(ctx context.Context, x string, emit func(string))\nfunc (f *myFn) OnTimer(ctx context.Context, t timer.Event)\n// after\nfunc (f *myFn) ProcessElement(ctx context.Context, x string, emit func(string))\nfunc (f *myFn) OnTimer(ctx context.Context, t timer.Event, emit func(string))","handlingStrategy":"validation","validationCode":"// Assert at build time that both methods either both have or both lack emit params.\nif hasEmits(fn.ProcessElement) != hasEmits(fn.OnTimer) { return errors.New(\"emitter existence mismatch\") }","typeGuard":"func hasEmits(method interface{}) bool { t := reflect.TypeOf(method); if t.Kind() == reflect.Func { for i := 0; i < t.NumIn(); i++ { if isEmitType(t.In(i)) { return true } } }; return false }","tryCatchPattern":"if err := beam.ParDo(scope, fn, input); err != nil { return fmt.Errorf(\"emitter mismatch: %w\", err) }","preventionTips":["Never give OnTimer a bare context-only signature if ProcessElement emits","Review DoFn signatures after refactors","Keep both methods adjacent in the source file"],"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"}