{"record":{"id":"f90eb4946e4d82a0","repo":"apache/beam","slug":"timer-using-dofn-v-doesn-t-use-a-kv-as-pcollection-input","errorCode":null,"errorMessage":"timer using DoFn %v doesn't use a KV as PCollection input. Unable to extract key coder for timers, got %v","messagePattern":"timer using DoFn (.+?) doesn't use a KV as PCollection input\\. Unable to extract key coder for timers, got (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/graphx/translate.go","lineNumber":632,"sourceCode":"\t\t\t\t\t\t},\n\t\t\t\t\t}\n\t\t\t\tdefault:\n\t\t\t\t\treturn nil, errors.Errorf(\"State type %v not recognized for state %v\", ps.StateKey(), ps)\n\t\t\t\t}\n\t\t\t}\n\t\t\tpayload.StateSpecs = stateSpecs\n\t\t}\n\t\tif _, ok := edge.Edge.DoFn.ProcessElementFn().TimerProvider(); ok {\n\t\t\tm.requirements[URNRequiresStatefulProcessing] = true\n\t\t\ttimerSpecs := make(map[string]*pipepb.TimerFamilySpec)\n\t\t\tpipelineTimers, _ := edge.Edge.DoFn.PipelineTimers()\n\n\t\t\t// All timers for a single DoFn have the same key and window coders, that match the input PCollection.\n\t\t\tmainInputID := inputs[\"i0\"]\n\t\t\tpCol := m.pcollections[mainInputID]\n\t\t\tkvCoder := m.coders.coders[pCol.CoderId]\n\t\t\tif kvCoder.GetSpec().GetUrn() != urnKVCoder {\n\t\t\t\treturn nil, errors.Errorf(\"timer using DoFn %v doesn't use a KV as PCollection input. Unable to extract key coder for timers, got %v\", edge.Name, kvCoder.GetSpec().GetUrn())\n\t\t\t}\n\t\t\tkeyCoderID := kvCoder.GetComponentCoderIds()[0]\n\n\t\t\twsID := pCol.GetWindowingStrategyId()\n\t\t\tws := m.windowing[wsID]\n\t\t\twindowCoderID := ws.GetWindowCoderId()\n\n\t\t\ttimerCoderID := m.coders.internBuiltInCoder(urnTimerCoder, keyCoderID, windowCoderID)\n\n\t\t\tfor _, pt := range pipelineTimers {\n\t\t\t\tfor timerFamilyID, timeDomain := range pt.Timers() {\n\t\t\t\t\ttimerSpecs[timerFamilyID] = &pipepb.TimerFamilySpec{\n\t\t\t\t\t\tTimeDomain:         pipepb.TimeDomain_Enum(timeDomain),\n\t\t\t\t\t\tTimerFamilyCoderId: timerCoderID,\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t}\n\t\t\tpayload.TimerFamilySpecs = timerSpecs","sourceCodeStart":614,"sourceCodeEnd":650,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/graphx/translate.go#L614-L650","documentation":"A DoFn that uses timers must read from a KV PCollection so the timer key coder can be extracted from the KV coder. When translating timers, the marshaller looked up the coder of the main input PCollection and found it is not a KV coder, so it cannot derive the key/window coders timers require, and fails naming the DoFn and the actual URN found.","triggerScenarios":"Applying a timer-using DoFn (implements TimerProvider) to a non-KV input PCollection, e.g. beam.ParDo(p, timerFn, pc) where pc's element type is not KV<K,V>.","commonSituations":"Forgetting beam.KV wrapper before a timer DoFn; restructurings/windowing that changed the input to a plain value; copy-pasting a stateful DoFn onto an unkeyed stream.","solutions":["Key the input before the timer DoFn: pass beam.KV{K: key, V: value} elements (use beam.ParDo with a KV output upstream)","Verify with pc.Type() that the input coder URN is urnKVCoder","If the DoFn only needs timers per element, introduce an artificial key (beam.AddDummyKey) then drop it after","Check for a restructure step that flattened the KV"],"exampleFix":"// before\nbeam.ParDo(s, &timerFn{}, in) // in is []string\n// after\nkeyed := beam.ParDo(s, addKeyFn, in) // emits KV<string,string>\nbeam.ParDo(s, &timerFn{}, keyed)","handlingStrategy":"validation","validationCode":"// Before applying a timer DoFn, assert the input is KV-typed\nif !beam.IsKV(in.Type()) {\n    return errors.New(\"timer DoFn requires a KV<...> PCollection input\")\n}","typeGuard":"func isKVCollection(pc beam.PCollection) bool {\n    return strings.HasPrefix(pc.Type().String(), \"beam.KV\")\n}","tryCatchPattern":"if _, err := graphx.Marshal(p); err != nil {\n    if strings.Contains(err.Error(), \"doesn't use a KV as PCollection input\") {\n        return fmt.Errorf(\"timer DoFn input must be KV: %v\", err)\n    }\n    return err\n}","preventionTips":["Insert beam.AddDummyKey (or a real keying step) before timer-using DoFns","Check PCollection types in pipeline tests before submission to runners","Annotate timer DoFns with comments stating KV input requirement"],"tags":["go","apache-beam","timers","stateful","kv"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}