{"record":{"id":"9b70cd6f1d0dfe1d","repo":"apache/beam","slug":"processelement-uses-a-timerprovider-but-is-not-keyed-all","errorCode":null,"errorMessage":"ProcessElement uses a TimerProvider, but is not keyed. All stateful DoFns must take a key/value pair as an input.","messagePattern":"ProcessElement uses a TimerProvider, but is not keyed\\. All stateful DoFns must take a key/value pair as an input\\.","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1424,"sourceCode":"\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.\")\n\t\t}\n\t\ttimerKeys := make(map[string]string)\n\t\tfor i, t := range pt {\n\t\t\tfor timerFamilyID := range t.Timers() {\n\t\t\t\tif timer, ok := timerKeys[timerFamilyID]; ok {\n\t\t\t\t\terr := errors.Errorf(\"Duplicate timer key %v\", timerFamilyID)\n\t\t\t\t\treturn errors.SetTopLevelMsgf(err, \"Duplicate timer family ID %v used by struct fields %v and %v. Ensure that timer family IDs are unique per DoFn\", timerFamilyID, timer, fieldNames[i])\n\t\t\t\t}\n\t\t\t\ttimerKeys[timerFamilyID] = fieldNames[i]\n\t\t\t}\n\t\t}","sourceCodeStart":1406,"sourceCodeEnd":1442,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1406-L1442","documentation":"Like state, timers in Beam Go are keyed: each timer belongs to a key so expiry events can be routed per key. This error is thrown when ProcessElement declares a TimerProvider parameter but the DoFn's input is a single (unkeyed) element, making it impossible to scope timers.","triggerScenarios":"Declaring a timer.Provider parameter in ProcessElement while the input element is not a KV pair, e.g. ProcessElement(ctx, tp timer.Provider, line string).","commonSituations":"Adding timers to an existing unkeyed DoFn; copying timer example code into a pipeline whose PCollections aren't keyed; forgetting the beam.AddFixedKey step.","solutions":["Key the input: change ProcessElement to accept a KV[K, V] and key the upstream PCollection (e.g. beam.AddFixedKey).","Remove the TimerProvider parameter if timers aren't needed.","Confirm the PCollection feeding this DoFn is of KV type."],"exampleFix":"// before\nfunc (fn *BatchFn) ProcessElement(ctx context.Context, tp timer.Provider, line string) error { ... }\n// after\nfunc (fn *BatchFn) ProcessElement(ctx context.Context, tp timer.Provider, k beam.KV[string, string]) error { ... }\n// pipe: beam.AddFixedKey(s, input)","handlingStrategy":"validation","validationCode":"// ensure the input to timer DoFns is keyed\nif !isKeyedPCollection(input) {\n    input = beam.AddFixedKey(s, input)\n}","typeGuard":"func isKeyedElem(v interface{}) bool {\n    switch v.(type) {\n    case beam.KV[interface{}, interface{}]:\n        return true\n    }\n    return false\n}","tryCatchPattern":"if err := beam.Run(ctx, p); err != nil {\n    if strings.Contains(err.Error(), \"uses a TimerProvider, but is not keyed\") {\n        log.Fatalf(\"timer DoFn %T requires a KV input: %v\", dofn, err)\n    }\n    return err\n}","preventionTips":["Key PCollections (beam.AddFixedKey or natural keys) before timer-based DoFns","Pair TimerProvider usage with KV inputs by convention","Validate DoFn signatures in unit tests before launching pipelines"],"tags":["go","apache-beam","dofn","timers","keyed-input"],"backgroundTag":"missing-required-argument","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"}